robustness: reduce memory consumption (#1155)

Detailed changes:
* Implement Kopia API client
* Implement kopia client
* Implement new persister
* Separate storeLoad and Delete concurrency tests
* Update Store interface to return error for Delete
* Return errors from os.RemoveAll
* Robustness test updates
* Push robustness metadata without writing to fs
* Fix testenv.AssertNoError references
* Update tests to use new kopia persister
* Minor updates to multiclient test cases
* Use require.NoError instead of assertNoError
* Add context to store interface
* Update logging to be less verbose
* Use io instead of ioutil package
* Simplify restore by using object ID
* Accommodate repository.NewWriter signature change
* Improve tests to increase code coverage
* Spelling and error string fixes
* Address lint errors

Co-authored-by: Nick <nick@kasten.io>

tests/robustness/checker/checker.go

@@ -98,11 +98,11 @@ func (ssMeta *SnapshotMetadata) IsDeleted() bool {
 }
 
 // GetSnapshotMetadata gets the metadata associated with the given snapshot ID.
-func (chk *Checker) GetSnapshotMetadata(snapID string) (*SnapshotMetadata, error) {
+func (chk *Checker) GetSnapshotMetadata(ctx context.Context, snapID string) (*SnapshotMetadata, error) {
 	chk.mu.RLock()
 	defer chk.mu.RUnlock()
 
-	return chk.loadSnapshotMetadata(snapID)
+	return chk.loadSnapshotMetadata(ctx, snapID)
 }
 
 // GetLiveSnapIDs gets the list of snapshot IDs being tracked by the checker's snapshot store
@@ -116,8 +116,8 @@ func (chk *Checker) GetLiveSnapIDs() []string {
 
 // IsSnapshotIDDeleted reports whether the metadata associated with the provided snapshot ID
 // has it marked as deleted.
-func (chk *Checker) IsSnapshotIDDeleted(snapID string) (bool, error) {
-	md, err := chk.loadSnapshotMetadata(snapID)
+func (chk *Checker) IsSnapshotIDDeleted(ctx context.Context, snapID string) (bool, error) {
+	md, err := chk.loadSnapshotMetadata(ctx, snapID)
 	if err != nil {
 		return false, err
 	}
@@ -157,7 +157,7 @@ func (chk *Checker) VerifySnapshotMetadata(ctx context.Context) error {
 
 			if chk.RecoveryMode {
 				chk.mu.Lock()
-				chk.snapshotMetadataStore.Delete(metaSnapID)
+				chk.snapshotMetadataStore.Delete(ctx, metaSnapID)
 				chk.SnapIDIndex.RemoveFromIndex(metaSnapID, LiveSnapshotsIdxName)
 				chk.mu.Unlock()
 			} else {
@@ -222,7 +222,7 @@ func (chk *Checker) TakeSnapshot(ctx context.Context, sourceDir string, opts map
 	chk.mu.Lock()
 	defer chk.mu.Unlock()
 
-	err = chk.saveSnapshotMetadata(ssMeta)
+	err = chk.saveSnapshotMetadata(ctx, ssMeta)
 	if err != nil {
 		return snapID, err
 	}
@@ -252,7 +252,7 @@ func (chk *Checker) RestoreSnapshot(ctx context.Context, snapID string, reportOu
 // using the Checker's Snapshotter, and performs a data consistency check on the
 // resulting tree using the saved snapshot data.
 func (chk *Checker) RestoreSnapshotToPath(ctx context.Context, snapID, destPath string, reportOut io.Writer, opts map[string]string) error {
-	ssMeta, err := chk.safeRestorePrepare(snapID)
+	ssMeta, err := chk.safeRestorePrepare(ctx, snapID)
 	if err != nil {
 		return err
 	}
@@ -262,7 +262,7 @@ func (chk *Checker) RestoreSnapshotToPath(ctx context.Context, snapID, destPath
 
 // safeRestorePrepare wraps key indexing operations that need to be
 // executed together under one mutex lock.
-func (chk *Checker) safeRestorePrepare(snapID string) (*SnapshotMetadata, error) {
+func (chk *Checker) safeRestorePrepare(ctx context.Context, snapID string) (*SnapshotMetadata, error) {
 	chk.mu.RLock()
 	defer chk.mu.RUnlock()
 
@@ -273,7 +273,7 @@ func (chk *Checker) safeRestorePrepare(snapID string) (*SnapshotMetadata, error)
 		return nil, robustness.ErrNoOp
 	}
 
-	ssMeta, err := chk.loadSnapshotMetadata(snapID)
+	ssMeta, err := chk.loadSnapshotMetadata(ctx, snapID)
 	if err != nil {
 		return nil, err
 	}
@@ -309,7 +309,7 @@ func (chk *Checker) RestoreVerifySnapshot(ctx context.Context, snapID, destPath
 	chk.mu.Lock()
 	defer chk.mu.Unlock()
 
-	err = chk.saveSnapshotMetadata(ssMeta)
+	err = chk.saveSnapshotMetadata(ctx, ssMeta)
 	if err != nil {
 		return err
 	}
@@ -333,7 +333,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[
 	// Load the metadata, ensure the snapshot hasn't already been deleted.
 	// If not, remove it from the list of live snapshots. These two tasks must
 	// be done atomically.
-	ssMeta, err := chk.safeDeletePrepare(snapID)
+	ssMeta, err := chk.safeDeletePrepare(ctx, snapID)
 	if err != nil {
 		return err
 	}
@@ -346,7 +346,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[
 	ssMeta.DeletionTime = clock.Now()
 	ssMeta.ValidationData = nil
 
-	return chk.safeDeleteFinish(ssMeta)
+	return chk.safeDeleteFinish(ctx, ssMeta)
 }
 
 // safeDeletePrepare is a routine that will remove the snapshot ID from
@@ -354,7 +354,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[
 // The check for deletion must be done atomically with the removal from the
 // index to ensure no races, which could lead to false failures if the Delete
 // operation runs against a snapshot that has already been deleted.
-func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error) {
+func (chk *Checker) safeDeletePrepare(ctx context.Context, snapID string) (*SnapshotMetadata, error) {
 	chk.mu.Lock()
 	defer chk.mu.Unlock()
 
@@ -364,7 +364,7 @@ func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error)
 		return nil, robustness.ErrNoOp
 	}
 
-	ssMeta, err := chk.loadSnapshotMetadata(snapID)
+	ssMeta, err := chk.loadSnapshotMetadata(ctx, snapID)
 	if err != nil {
 		return nil, err
 	}
@@ -381,13 +381,13 @@ func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error)
 
 // safeDelete finish will atomically update the metadata repository with the
 // deleted snapshot metadata and add the ID to the deleted snapshots index.
-func (chk *Checker) safeDeleteFinish(ssMeta *SnapshotMetadata) error {
+func (chk *Checker) safeDeleteFinish(ctx context.Context, ssMeta *SnapshotMetadata) error {
 	chk.mu.Lock()
 	defer chk.mu.Unlock()
 
 	chk.SnapIDIndex.AddToIndex(ssMeta.SnapID, DeletedSnapshotsIdxName)
 
-	err := chk.saveSnapshotMetadata(ssMeta)
+	err := chk.saveSnapshotMetadata(ctx, ssMeta)
 	if err != nil {
 		return err
 	}
@@ -395,13 +395,13 @@ func (chk *Checker) safeDeleteFinish(ssMeta *SnapshotMetadata) error {
 	return nil
 }
 
-func (chk *Checker) saveSnapshotMetadata(ssMeta *SnapshotMetadata) error {
+func (chk *Checker) saveSnapshotMetadata(ctx context.Context, ssMeta *SnapshotMetadata) error {
 	ssMetaRaw, err := json.Marshal(ssMeta)
 	if err != nil {
 		return err
 	}
 
-	err = chk.snapshotMetadataStore.Store(ssMeta.SnapID, ssMetaRaw)
+	err = chk.snapshotMetadataStore.Store(ctx, ssMeta.SnapID, ssMetaRaw)
 	if err != nil {
 		return err
 	}
@@ -409,9 +409,9 @@ func (chk *Checker) saveSnapshotMetadata(ssMeta *SnapshotMetadata) error {
 	return nil
 }
 
-func (chk *Checker) loadSnapshotMetadata(snapID string) (*SnapshotMetadata, error) {
+func (chk *Checker) loadSnapshotMetadata(ctx context.Context, snapID string) (*SnapshotMetadata, error) {
 	// Lookup metadata by snapshot ID
-	b, err := chk.snapshotMetadataStore.Load(snapID)
+	b, err := chk.snapshotMetadataStore.Load(ctx, snapID)
 	if err != nil {
 		return nil, err
 	}

tests/robustness/engine/engine.go

@@ -134,17 +134,17 @@ func (e *Engine) Shutdown(ctx context.Context) error {
 	defer e.cleanComponents()
 
 	if e.MetaStore != nil {
-		err := e.saveLog()
+		err := e.saveLog(ctx)
 		if err != nil {
 			return err
 		}
 
-		err = e.saveStats()
+		err = e.saveStats(ctx)
 		if err != nil {
 			return err
 		}
 
-		err = e.saveSnapIDIndex()
+		err = e.saveSnapIDIndex(ctx)
 		if err != nil {
 			return err
 		}
@@ -193,19 +193,19 @@ func (e *Engine) Init(ctx context.Context) error {
 		return err
 	}
 
-	err = e.loadStats()
+	err = e.loadStats(ctx)
 	if err != nil {
 		return err
 	}
 
 	e.CumulativeStats.RunCounter++
 
-	err = e.loadLog()
+	err = e.loadLog(ctx)
 	if err != nil {
 		return err
 	}
 
-	err = e.loadSnapIDIndex()
+	err = e.loadSnapIDIndex(ctx)
 	if err != nil {
 		return err
 	}

tests/robustness/engine/engine_test.go

@@ -298,7 +298,7 @@ func TestSnapshotVerificationFail(t *testing.T) {
 	require.NoError(t, err)
 
 	// Get the metadata collected on that snapshot
-	ssMeta1, err := eng.Checker.GetSnapshotMetadata(snapID1)
+	ssMeta1, err := eng.Checker.GetSnapshotMetadata(ctx, snapID1)
 	require.NoError(t, err)
 
 	// Do additional writes, writing 1 extra byte than before
@@ -310,7 +310,7 @@ func TestSnapshotVerificationFail(t *testing.T) {
 	require.NoError(t, err)
 
 	// Get the second snapshot's metadata
-	ssMeta2, err := eng.Checker.GetSnapshotMetadata(snapID2)
+	ssMeta2, err := eng.Checker.GetSnapshotMetadata(ctx, snapID2)
 	require.NoError(t, err)
 
 	// Swap second snapshot's validation data into the first's metadata
@@ -369,11 +369,11 @@ func TestDataPersistency(t *testing.T) {
 	require.NoError(t, err)
 
 	// Get the walk data associated with the snapshot that was taken
-	dataDirWalk, err := eng.Checker.GetSnapshotMetadata(snapID)
+	dataDirWalk, err := eng.Checker.GetSnapshotMetadata(ctx, snapID)
 	require.NoError(t, err)
 
 	// Save the snapshot ID index
-	err = eng.saveSnapIDIndex()
+	err = eng.saveSnapIDIndex(ctx)
 	require.NoError(t, err)
 
 	// Flush the snapshot metadata to persistent storage
@@ -693,6 +693,8 @@ func TestIOLimitPerWriteAction(t *testing.T) {
 }
 
 func TestStatsPersist(t *testing.T) {
+	ctx := context.Background()
+
 	tmpDir, err := ioutil.TempDir("", "stats-persist-test")
 	require.NoError(t, err)
 
@@ -729,7 +731,7 @@ func TestStatsPersist(t *testing.T) {
 		},
 	}
 
-	err = eng.saveStats()
+	err = eng.saveStats(ctx)
 	require.NoError(t, err)
 
 	err = eng.MetaStore.FlushMetadata()
@@ -749,7 +751,7 @@ func TestStatsPersist(t *testing.T) {
 		MetaStore: snapStoreNew,
 	}
 
-	err = engNew.loadStats()
+	err = engNew.loadStats(ctx)
 	require.NoError(t, err)
 
 	if got, want := engNew.Stats(), eng.Stats(); got != want {
@@ -761,6 +763,8 @@ func TestStatsPersist(t *testing.T) {
 }
 
 func TestLogsPersist(t *testing.T) {
+	ctx := context.Background()
+
 	tmpDir, err := ioutil.TempDir("", "logs-persist-test")
 	require.NoError(t, err)
 
@@ -799,7 +803,7 @@ func TestLogsPersist(t *testing.T) {
 		EngineLog: logData,
 	}
 
-	err = eng.saveLog()
+	err = eng.saveLog(ctx)
 	require.NoError(t, err)
 
 	err = eng.MetaStore.FlushMetadata()
@@ -819,7 +823,7 @@ func TestLogsPersist(t *testing.T) {
 		MetaStore: snapStoreNew,
 	}
 
-	err = engNew.loadLog()
+	err = engNew.loadLog(ctx)
 	require.NoError(t, err)
 
 	if got, want := engNew.EngineLog.String(), eng.EngineLog.String(); got != want {

tests/robustness/engine/metadata.go

@@ -3,6 +3,7 @@
 package engine
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 
@@ -17,18 +18,18 @@ const (
 )
 
 // saveLog saves the engine Log in the metadata store.
-func (e *Engine) saveLog() error {
+func (e *Engine) saveLog(ctx context.Context) error {
 	b, err := json.Marshal(e.EngineLog)
 	if err != nil {
 		return err
 	}
 
-	return e.MetaStore.Store(engineLogsStoreKey, b)
+	return e.MetaStore.Store(ctx, engineLogsStoreKey, b)
 }
 
 // loadLog loads the engine log from the metadata store.
-func (e *Engine) loadLog() error {
-	b, err := e.MetaStore.Load(engineLogsStoreKey)
+func (e *Engine) loadLog(ctx context.Context) error {
+	b, err := e.MetaStore.Load(ctx, engineLogsStoreKey)
 	if err != nil {
 		if errors.Is(err, robustness.ErrKeyNotFound) {
 			// Swallow key-not-found error. May not have historical logs
@@ -49,18 +50,18 @@ func (e *Engine) loadLog() error {
 }
 
 // saveStats saves the engine Stats in the metadata store.
-func (e *Engine) saveStats() error {
+func (e *Engine) saveStats(ctx context.Context) error {
 	cumulStatRaw, err := json.Marshal(e.CumulativeStats)
 	if err != nil {
 		return err
 	}
 
-	return e.MetaStore.Store(engineStatsStoreKey, cumulStatRaw)
+	return e.MetaStore.Store(ctx, engineStatsStoreKey, cumulStatRaw)
 }
 
 // loadStats loads the engine Stats from the metadata store.
-func (e *Engine) loadStats() error {
-	b, err := e.MetaStore.Load(engineStatsStoreKey)
+func (e *Engine) loadStats(ctx context.Context) error {
+	b, err := e.MetaStore.Load(ctx, engineStatsStoreKey)
 	if err != nil {
 		if errors.Is(err, robustness.ErrKeyNotFound) {
 			// Swallow key-not-found error. We may not have historical
@@ -78,18 +79,18 @@ func (e *Engine) loadStats() error {
 }
 
 // saveSnapIDIndex saves the Checker's snapshot ID index in the metadata store.
-func (e *Engine) saveSnapIDIndex() error {
+func (e *Engine) saveSnapIDIndex(ctx context.Context) error {
 	snapIDIdxRaw, err := json.Marshal(e.Checker.SnapIDIndex)
 	if err != nil {
 		return err
 	}
 
-	return e.MetaStore.Store(snapIDIndexStoreKey, snapIDIdxRaw)
+	return e.MetaStore.Store(ctx, snapIDIndexStoreKey, snapIDIdxRaw)
 }
 
 // loadSnapIDIndex loads the Checker's snapshot ID index from the metadata store.
-func (e *Engine) loadSnapIDIndex() error {
-	b, err := e.MetaStore.Load(snapIDIndexStoreKey)
+func (e *Engine) loadSnapIDIndex(ctx context.Context) error {
+	b, err := e.MetaStore.Load(ctx, snapIDIndexStoreKey)
 	if err != nil {
 		if errors.Is(err, robustness.ErrKeyNotFound) {
 			// Swallow key-not-found error. We may not have historical

tests/robustness/multiclient_test/framework/harness.go

@@ -44,7 +44,7 @@ type TestHarness struct {
 	baseDirPath string
 	fileWriter  *MultiClientFileWriter
 	snapshotter *MultiClientSnapshotter
-	persister   *snapmeta.KopiaPersister
+	persister   *snapmeta.KopiaPersisterLight
 	engine      *engine.Engine
 
 	skipTest bool
@@ -149,16 +149,9 @@ func (th *TestHarness) getSnapshotter() bool {
 }
 
 func (th *TestHarness) getPersister() bool {
-	kp, err := snapmeta.NewPersister(th.baseDirPath)
+	kp, err := snapmeta.NewPersisterLight(th.baseDirPath)
 	if err != nil {
-		if errors.Is(err, kopiarunner.ErrExeVariableNotSet) {
-			log.Println("Skipping robustness tests because KOPIA_EXE is not set")
-
-			th.skipTest = true
-		} else {
-			log.Println("Error creating kopia Persister:", err)
-		}
-
+		log.Println("Error creating kopia Persister:", err)
 		return false
 	}
 

tests/robustness/multiclient_test/multiclient_test.go

@@ -44,14 +44,14 @@ func TestManySmallFiles(t *testing.T) {
 		_, err = eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts)
 		require.NoError(t, err)
 
-		_, err = eng.ExecAction(ctx, engine.SnapshotDirActionKey, nil)
+		snapOut, err := eng.ExecAction(ctx, engine.SnapshotDirActionKey, nil)
 		require.NoError(t, err)
 
-		_, err = eng.ExecAction(ctx, engine.RestoreSnapshotActionKey, nil)
+		_, err = eng.ExecAction(ctx, engine.RestoreSnapshotActionKey, snapOut)
 		require.NoError(t, err)
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 	th.RunN(ctx, t, numClients, f)
 }
 
@@ -82,7 +82,7 @@ func TestOneLargeFile(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 	th.RunN(ctx, t, numClients, f)
 }
 
@@ -117,12 +117,12 @@ func TestManySmallFilesAcrossDirecoryTree(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 	th.RunN(ctx, t, numClients, f)
 }
 
 func TestRandomizedSmall(t *testing.T) {
-	numClients := 2
+	numClients := 4
 	st := clock.Now()
 
 	opts := engine.ActionOpts{
@@ -151,7 +151,7 @@ func TestRandomizedSmall(t *testing.T) {
 		}
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 	th.RunN(ctx, t, numClients, f)
 }
 

tests/robustness/persister.go

@@ -2,12 +2,14 @@
 
 package robustness
 
+import "context"
+
 // Store describes the ability to store and retrieve
 // a buffer of metadata, indexed by a string key.
 type Store interface {
-	Store(key string, val []byte) error
-	Load(key string) ([]byte, error)
-	Delete(key string)
+	Store(ctx context.Context, key string, val []byte) error
+	Load(ctx context.Context, key string) ([]byte, error)
+	Delete(ctx context.Context, key string) error
 }
 
 // Persister describes the ability to flush metadata

tests/robustness/robustness_test/main_test.go

@@ -73,7 +73,7 @@ type kopiaRobustnessTestHarness struct {
 	baseDirPath string
 	fileWriter  *fiofilewriter.FileWriter
 	snapshotter *snapmeta.KopiaSnapshotter
-	persister   *snapmeta.KopiaPersister
+	persister   *snapmeta.KopiaPersisterLight
 	engine      *engine.Engine
 
 	skipTest bool
@@ -154,16 +154,9 @@ func (th *kopiaRobustnessTestHarness) getSnapshotter() bool {
 }
 
 func (th *kopiaRobustnessTestHarness) getPersister() bool {
-	kp, err := snapmeta.NewPersister(th.baseDirPath)
+	kp, err := snapmeta.NewPersisterLight(th.baseDirPath)
 	if err != nil {
-		if errors.Is(err, kopiarunner.ErrExeVariableNotSet) {
-			log.Println("Skipping robustness tests because KOPIA_EXE is not set")
-
-			th.skipTest = true
-		} else {
-			log.Println("Error creating kopia Persister:", err)
-		}
-
+		log.Println("Error creating kopia Persister:", err)
 		return false
 	}
 

tests/robustness/robustness_test/robustness_test.go

@@ -29,7 +29,7 @@ func TestManySmallFiles(t *testing.T) {
 		fiofilewriter.MinNumFilesPerWriteField: strconv.Itoa(numFiles),
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 
 	_, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts)
 	require.NoError(t, err)
@@ -53,7 +53,7 @@ func TestOneLargeFile(t *testing.T) {
 		fiofilewriter.MinNumFilesPerWriteField: strconv.Itoa(numFiles),
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 
 	_, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts)
 	require.NoError(t, err)
@@ -81,7 +81,7 @@ func TestManySmallFilesAcrossDirecoryTree(t *testing.T) {
 		engine.ActionRepeaterField:             strconv.Itoa(actionRepeats),
 	}
 
-	ctx := testlogging.Context(t)
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 
 	_, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts)
 	require.NoError(t, err)
@@ -112,9 +112,8 @@ func TestRandomizedSmall(t *testing.T) {
 		},
 	}
 
+	ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo)
 	for clock.Since(st) <= *randomizedTestDur {
-		ctx := testlogging.Context(t)
-
 		err := eng.RandomAction(ctx, opts)
 		if errors.Is(err, robustness.ErrNoOp) {
 			t.Log("Random action resulted in no-op")

tests/robustness/snapmeta/kopia_persister_light.go

@@ -0,0 +1,114 @@
+// +build darwin,amd64 linux,amd64
+
+package snapmeta
+
+import (
+	"context"
+	"log"
+	"os"
+	"sync"
+
+	"github.com/kopia/kopia/tests/robustness"
+	"github.com/kopia/kopia/tests/tools/kopiaclient"
+)
+
+// KopiaPersisterLight is a wrapper for KopiaClient that satisfies the Persister
+// interface.
+type KopiaPersisterLight struct {
+	kc            *kopiaclient.KopiaClient
+	keysInProcess map[string]bool
+	c             *sync.Cond
+	baseDir       string
+}
+
+var _ robustness.Persister = (*KopiaPersisterLight)(nil)
+
+// NewPersisterLight returns a new KopiaPersisterLight.
+func NewPersisterLight(baseDir string) (*KopiaPersisterLight, error) {
+	persistenceDir, err := os.MkdirTemp(baseDir, "kopia-persistence-root-")
+	if err != nil {
+		return nil, err
+	}
+
+	return &KopiaPersisterLight{
+		kc:            kopiaclient.NewKopiaClient(persistenceDir),
+		keysInProcess: map[string]bool{},
+		c:             sync.NewCond(&sync.Mutex{}),
+		baseDir:       persistenceDir,
+	}, nil
+}
+
+// ConnectOrCreateRepo creates a new Kopia repo or connects to an existing one if possible.
+func (kpl *KopiaPersisterLight) ConnectOrCreateRepo(repoPath string) error {
+	bucketName := os.Getenv(S3BucketNameEnvKey)
+	return kpl.kc.CreateOrConnectRepo(context.Background(), repoPath, bucketName)
+}
+
+// Store pushes the key value pair to the Kopia repository.
+func (kpl *KopiaPersisterLight) Store(ctx context.Context, key string, val []byte) error {
+	kpl.waitFor(key)
+	defer kpl.doneWith(key)
+
+	log.Println("pushing metadata for", key)
+
+	return kpl.kc.SnapshotCreate(ctx, key, val)
+}
+
+// Load pulls the key value pair from the Kopia repo and returns the value.
+func (kpl *KopiaPersisterLight) Load(ctx context.Context, key string) ([]byte, error) {
+	kpl.waitFor(key)
+	defer kpl.doneWith(key)
+
+	log.Println("pulling metadata for", key)
+
+	return kpl.kc.SnapshotRestore(ctx, key)
+}
+
+// Delete deletes all snapshots associated with the given key.
+func (kpl *KopiaPersisterLight) Delete(ctx context.Context, key string) error {
+	kpl.waitFor(key)
+	defer kpl.doneWith(key)
+
+	log.Println("deleting metadata for", key)
+
+	return kpl.kc.SnapshotDelete(ctx, key)
+}
+
+// LoadMetadata is a no-op. It is included to satisfy the Persister interface.
+func (kpl *KopiaPersisterLight) LoadMetadata() error {
+	return nil
+}
+
+// FlushMetadata is a no-op. It is included to satisfy the Persister interface.
+func (kpl *KopiaPersisterLight) FlushMetadata() error {
+	return nil
+}
+
+// GetPersistDir returns the persistence directory.
+func (kpl *KopiaPersisterLight) GetPersistDir() string {
+	return kpl.baseDir
+}
+
+// Cleanup removes the persistence directory and closes the Kopia repo.
+func (kpl *KopiaPersisterLight) Cleanup() {
+	if err := os.RemoveAll(kpl.baseDir); err != nil {
+		log.Println("cannot remove persistence dir")
+	}
+}
+
+func (kpl *KopiaPersisterLight) waitFor(key string) {
+	kpl.c.L.Lock()
+	for kpl.keysInProcess[key] {
+		kpl.c.Wait()
+	}
+
+	kpl.keysInProcess[key] = true
+	kpl.c.L.Unlock()
+}
+
+func (kpl *KopiaPersisterLight) doneWith(key string) {
+	kpl.c.L.Lock()
+	delete(kpl.keysInProcess, key)
+	kpl.c.L.Unlock()
+	kpl.c.Signal()
+}

tests/robustness/snapmeta/kopia_persister_light_test.go

@@ -0,0 +1,168 @@
+// +build darwin,amd64 linux,amd64
+
+package snapmeta
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"testing"
+)
+
+var (
+	key = "mykey"
+	val = []byte("myval")
+)
+
+func TestStoreLoadDelete(t *testing.T) {
+	ctx := context.Background()
+
+	repoPath, err := os.MkdirTemp("", "kopia-test-repo-")
+	assertNoError(t, err)
+
+	defer os.RemoveAll(repoPath)
+
+	kpl := initKPL(t, repoPath)
+	defer kpl.Cleanup()
+
+	kpl.testStoreLoad(ctx, t, key, val)
+	kpl.testDelete(ctx, t, key)
+}
+
+func TestConcurrency(t *testing.T) {
+	ctx := context.Background()
+
+	repoPath, err := os.MkdirTemp("", "kopia-test-repo-")
+	assertNoError(t, err)
+
+	defer os.RemoveAll(repoPath)
+
+	kpl := initKPL(t, repoPath)
+	defer kpl.Cleanup()
+
+	keys := []string{"key1", "key2", "key3"}
+	vals := [][]byte{[]byte("val1"), []byte("val2"), []byte("val3")}
+
+	t.Run("storeLoad", func(t *testing.T) {
+		for i := 0; i < 9; i++ {
+			j := i
+			t.Run(fmt.Sprint(i), func(t *testing.T) {
+				t.Parallel()
+				kpl.testStoreLoad(ctx, t, keys[j%3], vals[j%3])
+			})
+		}
+	})
+
+	t.Run("delete", func(t *testing.T) {
+		for i := 0; i < 9; i++ {
+			j := i
+			t.Run(fmt.Sprint(i), func(t *testing.T) {
+				t.Parallel()
+				kpl.testDelete(ctx, t, keys[j%3])
+			})
+		}
+	})
+}
+
+// Store and test that subsequent Load succeeds.
+func (kpl *KopiaPersisterLight) testStoreLoad(ctx context.Context, t *testing.T, key string, val []byte) { //nolint:thelper
+	err := kpl.Store(ctx, key, val)
+	assertNoError(t, err)
+
+	valOut, err := kpl.Load(ctx, key)
+	assertNoError(t, err)
+
+	if !bytes.Equal(valOut, val) {
+		t.Fatal("loaded value does not equal stored value", valOut, val)
+	}
+}
+
+// Delete and test that subsequent Load fails.
+func (kpl *KopiaPersisterLight) testDelete(ctx context.Context, t *testing.T, key string) { //nolint:thelper
+	kpl.Delete(ctx, key)
+
+	_, err := kpl.Load(ctx, key)
+	log.Println("err:", err)
+
+	if err == nil {
+		t.Fatal("snapshot was not deleted properly")
+	}
+}
+
+func TestPersistence(t *testing.T) {
+	ctx := context.Background()
+
+	repoPath, err := os.MkdirTemp("", "kopia-test-repo-")
+	assertNoError(t, err)
+
+	kpl := initKPL(t, repoPath)
+
+	// Persistence directory should be set.
+	if persistDir := kpl.GetPersistDir(); persistDir == "" {
+		t.Error("could not get persistence directory")
+	}
+
+	// These are no-ops and should always succeed.
+	err = kpl.LoadMetadata()
+	assertNoError(t, err)
+	err = kpl.FlushMetadata()
+	assertNoError(t, err)
+
+	// Store and cleanup kpl
+	err = kpl.Store(ctx, key, val)
+	assertNoError(t, err)
+
+	kpl.Cleanup()
+
+	// Re-initialize and Load
+	kpl = initKPL(t, repoPath)
+	valOut, err := kpl.Load(ctx, key)
+	assertNoError(t, err)
+
+	if !bytes.Equal(valOut, val) {
+		t.Fatal("loaded value does not equal stored value")
+	}
+
+	kpl.Cleanup()
+	os.RemoveAll(repoPath)
+}
+
+func TestS3Connect(t *testing.T) {
+	repoPath, err := os.MkdirTemp("", "kopia-test-repo-")
+	assertNoError(t, err)
+
+	// Test the S3 code path by attempting to connect to a nonexistent bucket.
+	os.Setenv(S3BucketNameEnvKey, "does-not-exist")
+
+	kpl, err := NewPersisterLight("")
+	assertNoError(t, err)
+
+	if err := kpl.ConnectOrCreateRepo(repoPath); err == nil {
+		t.Error("should not be able to connect to nonexistent S3 bucket")
+	}
+
+	kpl.Cleanup()
+	os.RemoveAll(repoPath)
+}
+
+func initKPL(t *testing.T, repoPath string) *KopiaPersisterLight { //nolint:thelper
+	os.Unsetenv(S3BucketNameEnvKey)
+
+	kpl, err := NewPersisterLight("")
+	assertNoError(t, err)
+
+	err = kpl.ConnectOrCreateRepo(repoPath)
+	assertNoError(t, err)
+
+	return kpl
+}
+
+func assertNoError(t *testing.T, err error) {
+	t.Helper()
+
+	if err != nil {
+		t.Errorf("err: %v", err)
+	}
+}

tests/robustness/snapmeta/simple.go

@@ -3,6 +3,8 @@
 package snapmeta
 
 import (
+	"context"
+
 	"github.com/kopia/kopia/tests/robustness"
 )
 
@@ -24,7 +26,7 @@ func NewSimple() *Simple {
 }
 
 // Store implements the Storer interface Store method.
-func (s *Simple) Store(key string, val []byte) error {
+func (s *Simple) Store(ctx context.Context, key string, val []byte) error {
 	buf := make([]byte, len(val))
 	_ = copy(buf, val)
 
@@ -34,7 +36,7 @@ func (s *Simple) Store(key string, val []byte) error {
 }
 
 // Load implements the Storer interface Load method.
-func (s *Simple) Load(key string) ([]byte, error) {
+func (s *Simple) Load(ctx context.Context, key string) ([]byte, error) {
 	if buf, found := s.Data[key]; found {
 		retBuf := make([]byte, len(buf))
 		_ = copy(retBuf, buf)
@@ -46,6 +48,8 @@ func (s *Simple) Load(key string) ([]byte, error) {
 }
 
 // Delete implements the Storer interface Delete method.
-func (s *Simple) Delete(key string) {
+func (s *Simple) Delete(ctx context.Context, key string) error {
 	delete(s.Data, key)
+
+	return nil
 }

tests/robustness/snapmeta/simple_test.go

@@ -4,6 +4,7 @@ package snapmeta
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"testing"
 
@@ -11,9 +12,11 @@ import (
 )
 
 func TestSimpleBasic(t *testing.T) {
+	ctx := context.Background()
+
 	simple := NewSimple()
 
-	gotData, err := simple.Load("non-existent-key")
+	gotData, err := simple.Load(ctx, "non-existent-key")
 	if !errors.Is(err, robustness.ErrKeyNotFound) {
 		t.Fatalf("Did not get expected error: %q", err)
 	}
@@ -24,9 +27,9 @@ func TestSimpleBasic(t *testing.T) {
 
 	storeKey := "key-to-store"
 	data := []byte("some stored data")
-	simple.Store(storeKey, data)
+	simple.Store(ctx, storeKey, data)
 
-	gotData, err = simple.Load(storeKey)
+	gotData, err = simple.Load(ctx, storeKey)
 	if err != nil {
 		t.Fatalf("Error getting key: %v", err)
 	}
@@ -35,9 +38,9 @@ func TestSimpleBasic(t *testing.T) {
 		t.Fatalf("Did not get the correct data")
 	}
 
-	simple.Delete(storeKey)
+	simple.Delete(ctx, storeKey)
 
-	gotData, err = simple.Load(storeKey)
+	gotData, err = simple.Load(ctx, storeKey)
 	if !errors.Is(err, robustness.ErrKeyNotFound) {
 		t.Fatalf("Did not get expected error: %q", err)
 	}

tests/tools/kopiaclient/kopiaclient.go

@@ -0,0 +1,248 @@
+// +build darwin,amd64 linux,amd64
+
+// Package kopiaclient provides a client to interact with a Kopia repo.
+package kopiaclient
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"log"
+	"os"
+	"path/filepath"
+
+	"github.com/pkg/errors"
+
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/fs/virtualfs"
+	"github.com/kopia/kopia/internal/units"
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/filesystem"
+	"github.com/kopia/kopia/repo/blob/s3"
+	"github.com/kopia/kopia/snapshot"
+	"github.com/kopia/kopia/snapshot/policy"
+	"github.com/kopia/kopia/snapshot/snapshotfs"
+	"github.com/kopia/kopia/tests/robustness"
+)
+
+// KopiaClient uses a Kopia repo to create, restore, and delete snapshots.
+type KopiaClient struct {
+	configPath string
+	pw         string
+}
+
+const (
+	configFileName           = "config"
+	password                 = "kj13498po&_EXAMPLE" //nolint:gosec
+	s3Endpoint               = "s3.amazonaws.com"
+	awsAccessKeyIDEnvKey     = "AWS_ACCESS_KEY_ID"
+	awsSecretAccessKeyEnvKey = "AWS_SECRET_ACCESS_KEY" //nolint:gosec
+	dataFileName             = "data"
+)
+
+// NewKopiaClient returns a new KopiaClient.
+func NewKopiaClient(basePath string) *KopiaClient {
+	return &KopiaClient{
+		configPath: filepath.Join(basePath, configFileName),
+		pw:         password,
+	}
+}
+
+// CreateOrConnectRepo creates a new Kopia repo or connects to an existing one if possible.
+func (kc *KopiaClient) CreateOrConnectRepo(ctx context.Context, repoDir, bucketName string) error {
+	st, err := kc.getStorage(ctx, repoDir, bucketName)
+	if err != nil {
+		return err
+	}
+
+	if iErr := repo.Initialize(ctx, st, &repo.NewRepositoryOptions{}, kc.pw); iErr != nil {
+		if !errors.Is(iErr, repo.ErrAlreadyInitialized) {
+			return errors.Wrap(iErr, "repo is already initialized")
+		}
+
+		log.Println("connecting to existing repository")
+	}
+
+	if iErr := repo.Connect(ctx, kc.configPath, st, kc.pw, &repo.ConnectOptions{}); iErr != nil {
+		return errors.Wrap(iErr, "error connecting to repository")
+	}
+
+	return errors.Wrap(err, "unable to open repository")
+}
+
+// SnapshotCreate creates a snapshot for the given path.
+func (kc *KopiaClient) SnapshotCreate(ctx context.Context, key string, val []byte) error {
+	r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{})
+	if err != nil {
+		return errors.Wrap(err, "cannot open repository")
+	}
+
+	ctx, rw, err := r.NewWriter(ctx, repo.WriteSessionOptions{})
+	if err != nil {
+		return errors.Wrap(err, "cannot get new repository writer")
+	}
+
+	si := kc.getSourceInfoFromKey(r, key)
+
+	policyTree, err := policy.TreeForSource(ctx, r, si)
+	if err != nil {
+		return errors.Wrap(err, "cannot get policy tree for source")
+	}
+
+	source := kc.getSourceForKeyVal(key, val)
+	u := snapshotfs.NewUploader(rw)
+
+	man, err := u.Upload(ctx, source, policyTree, si)
+	if err != nil {
+		return errors.Wrap(err, "cannot get manifest")
+	}
+
+	log.Printf("snapshotting %v", units.BytesStringBase10(man.Stats.TotalFileSize))
+
+	if _, err := snapshot.SaveSnapshot(ctx, rw, man); err != nil {
+		return errors.Wrap(err, "cannot save snapshot")
+	}
+
+	if err := rw.Flush(ctx); err != nil {
+		return err
+	}
+
+	return r.Close(ctx)
+}
+
+// SnapshotRestore restores the latest snapshot for the given path.
+func (kc *KopiaClient) SnapshotRestore(ctx context.Context, key string) ([]byte, error) {
+	r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{})
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot open repository")
+	}
+
+	mans, err := kc.getSnapshotsFromKey(ctx, r, key)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot get snapshots from key")
+	}
+
+	man := kc.latestManifest(mans)
+	rootOIDWithPath := string(man.RootObjectID()) + "/" + dataFileName
+
+	oid, err := snapshotfs.ParseObjectIDWithPath(ctx, r, rootOIDWithPath)
+	if err != nil {
+		return nil, errors.Wrapf(err, "cannot parse object ID %s", rootOIDWithPath)
+	}
+
+	or, err := r.OpenObject(ctx, oid)
+	if err != nil {
+		return nil, errors.Wrapf(err, "cannot open object %s", string(oid))
+	}
+
+	val, err := io.ReadAll(or)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Printf("restored %v", units.BytesStringBase10(int64(len(val))))
+
+	if err := r.Close(ctx); err != nil {
+		return nil, err
+	}
+
+	return val, nil
+}
+
+// SnapshotDelete deletes all snapshots for a given path.
+func (kc *KopiaClient) SnapshotDelete(ctx context.Context, key string) error {
+	r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{})
+	if err != nil {
+		return errors.Wrap(err, "cannot open repository")
+	}
+
+	ctx, rw, err := r.NewWriter(ctx, repo.WriteSessionOptions{})
+	if err != nil {
+		return errors.Wrap(err, "cannot get new repository writer")
+	}
+
+	mans, err := kc.getSnapshotsFromKey(ctx, r, key)
+	if err != nil {
+		return errors.Wrap(err, "cannot get snapshots from key")
+	}
+
+	for _, man := range mans {
+		if err := rw.DeleteManifest(ctx, man.ID); err != nil {
+			return errors.Wrap(err, "cannot delete manifest")
+		}
+	}
+
+	if err := rw.Flush(ctx); err != nil {
+		return err
+	}
+
+	return r.Close(ctx)
+}
+
+func (kc *KopiaClient) getStorage(ctx context.Context, repoDir, bucketName string) (st blob.Storage, err error) {
+	if bucketName != "" {
+		s3Opts := &s3.Options{
+			BucketName:      bucketName,
+			Prefix:          repoDir,
+			Endpoint:        s3Endpoint,
+			AccessKeyID:     os.Getenv(awsAccessKeyIDEnvKey),
+			SecretAccessKey: os.Getenv(awsSecretAccessKeyEnvKey),
+		}
+		st, err = s3.New(ctx, s3Opts)
+	} else {
+		if iErr := os.MkdirAll(repoDir, 0o700); iErr != nil {
+			return nil, errors.Wrap(iErr, "cannot create directory")
+		}
+
+		fsOpts := &filesystem.Options{
+			Path: repoDir,
+		}
+		st, err = filesystem.New(ctx, fsOpts)
+	}
+
+	return st, errors.Wrap(err, "unable to get storage")
+}
+
+// getSourceForKeyVal creates a virtual directory for `key` that contains a single virtual file that
+// reads its contents from `val`.
+func (kc *KopiaClient) getSourceForKeyVal(key string, val []byte) fs.Entry {
+	return virtualfs.NewStaticDirectory(key, fs.Entries{
+		virtualfs.StreamingFileFromReader(dataFileName, bytes.NewReader(val)),
+	})
+}
+
+func (kc *KopiaClient) getSnapshotsFromKey(ctx context.Context, r repo.Repository, key string) ([]*snapshot.Manifest, error) {
+	si := kc.getSourceInfoFromKey(r, key)
+
+	manifests, err := snapshot.ListSnapshots(ctx, r, si)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot list snapshots")
+	}
+
+	if len(manifests) == 0 {
+		return nil, robustness.ErrKeyNotFound
+	}
+
+	return manifests, nil
+}
+
+func (kc *KopiaClient) getSourceInfoFromKey(r repo.Repository, key string) snapshot.SourceInfo {
+	return snapshot.SourceInfo{
+		Host:     r.ClientOptions().Hostname,
+		UserName: r.ClientOptions().Username,
+		Path:     key,
+	}
+}
+
+func (kc *KopiaClient) latestManifest(mans []*snapshot.Manifest) *snapshot.Manifest {
+	latest := mans[0]
+
+	for _, m := range mans {
+		if m.StartTime.After(latest.StartTime) {
+			latest = m
+		}
+	}
+
+	return latest
+}