Allow deleting batches of models/manifests (#4021)

Take a vararg of manifests to delete and delete them all in a single write session. This should help reduce the number of manifest blobs that kopia creates since it will only flush the pending manifest entries once during the operation.

---

#### 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

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

#### Issue(s)

* #4019

#### Test Plan

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

src/internal/kopia/model_store.go

@@ -497,20 +497,32 @@ func (ms *ModelStore) Delete(ctx context.Context, s model.Schema, id model.Stabl
 		return err
 	}
 
-	return ms.DeleteWithModelStoreID(ctx, latest)
+	return ms.DeleteWithModelStoreIDs(ctx, latest)
 }
 
-// DeleteWithModelStoreID deletes the model with the given ModelStoreID from the
-// model store. Turns into a noop if id is not empty but the model does not
-// exist.
-func (ms *ModelStore) DeleteWithModelStoreID(ctx context.Context, id manifest.ID) error {
-	if len(id) == 0 {
-		return clues.Stack(errNoModelStoreID).WithClues(ctx)
-	}
-
+// DeleteWithModelStoreID deletes the model(s) with the given ModelStoreID(s)
+// from the model store. For an individual ID, turns into a noop if the ID is
+// non-empty but the model doesn't exist. All model deletions should be
+// persisted atomically.
+//
+// Will not make any changes if any deletion attempt returns an error.
+func (ms *ModelStore) DeleteWithModelStoreIDs(
+	ctx context.Context,
+	ids ...manifest.ID,
+) error {
 	opts := repo.WriteSessionOptions{Purpose: "ModelStoreDelete"}
 	cb := func(innerCtx context.Context, w repo.RepositoryWriter) error {
-		return w.DeleteManifest(innerCtx, id)
+		for _, id := range ids {
+			if len(id) == 0 {
+				return clues.Stack(errNoModelStoreID).WithClues(ctx)
+			}
+
+			if err := w.DeleteManifest(innerCtx, id); err != nil {
+				return clues.Stack(err).WithClues(innerCtx).With("model_store_id", id)
+			}
+		}
+
+		return nil
 	}
 
 	if err := repo.WriteSession(ctx, ms.c, opts, cb); err != nil {

src/internal/kopia/model_store_test.go

@@ -169,7 +169,7 @@ func (suite *ModelStoreIntegrationSuite) TestNoIDsErrors() {
 	err = suite.m.Delete(suite.ctx, theModelType, "")
 	assert.Error(t, err, clues.ToCore(err))
 
-	err = suite.m.DeleteWithModelStoreID(suite.ctx, "")
+	err = suite.m.DeleteWithModelStoreIDs(suite.ctx, "")
 	assert.Error(t, err, clues.ToCore(err))
 }
 
@@ -711,13 +711,37 @@ func (suite *ModelStoreIntegrationSuite) TestPutDelete() {
 	assert.ErrorIs(t, err, data.ErrNotFound, clues.ToCore(err))
 }
 
+func (suite *ModelStoreIntegrationSuite) TestPutDeleteBatch() {
+	t := suite.T()
+	theModelType := model.BackupOpSchema
+	ids := []manifest.ID{}
+
+	for i := 0; i < 5; i++ {
+		foo := &fooModel{Bar: uuid.NewString()}
+
+		err := suite.m.Put(suite.ctx, theModelType, foo)
+		require.NoError(t, err, clues.ToCore(err))
+
+		ids = append(ids, foo.ModelStoreID)
+	}
+
+	err := suite.m.DeleteWithModelStoreIDs(suite.ctx, ids...)
+	require.NoError(t, err, clues.ToCore(err))
+
+	for _, id := range ids {
+		returned := &fooModel{}
+		err := suite.m.GetWithModelStoreID(suite.ctx, theModelType, id, returned)
+		assert.ErrorIs(t, err, data.ErrNotFound, clues.ToCore(err))
+	}
+}
+
 func (suite *ModelStoreIntegrationSuite) TestPutDelete_BadIDsNoop() {
 	t := suite.T()
 
 	err := suite.m.Delete(suite.ctx, model.BackupOpSchema, "foo")
 	assert.NoError(t, err, clues.ToCore(err))
 
-	err = suite.m.DeleteWithModelStoreID(suite.ctx, "foo")
+	err = suite.m.DeleteWithModelStoreIDs(suite.ctx, "foo")
 	assert.NoError(t, err, clues.ToCore(err))
 }
 

src/pkg/store/backup.go

@@ -61,7 +61,7 @@ type (
 
 	Storer interface {
 		Delete(ctx context.Context, s model.Schema, id model.StableID) error
-		DeleteWithModelStoreID(ctx context.Context, id manifest.ID) error
+		DeleteWithModelStoreIDs(ctx context.Context, ids ...manifest.ID) error
 		Get(ctx context.Context, s model.Schema, id model.StableID, data model.Model) error
 		GetIDsForType(ctx context.Context, s model.Schema, tags map[string]string) ([]*model.BaseModel, error)
 		GetWithModelStoreID(ctx context.Context, s model.Schema, id manifest.ID, data model.Model) error

src/pkg/store/mock/model_store.go

@@ -34,7 +34,7 @@ func (mms *ModelStore) Delete(ctx context.Context, s model.Schema, id model.Stab
 	return mms.err
 }
 
-func (mms *ModelStore) DeleteWithModelStoreID(ctx context.Context, id manifest.ID) error {
+func (mms *ModelStore) DeleteWithModelStoreIDs(ctx context.Context, ids ...manifest.ID) error {
 	return mms.err
 }