use count bus in kopia backups (#4482)

uses the count bus in the kopia backup package.
This currently duplicates counts that we're getting
from the kopia stats.  A later pr will remove the old
stats entirely in favor of the counter.

---

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

- [x] ⛔ No

#### Type of change

- [x] 🧹 Tech Debt/Cleanup

#### Test Plan

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

src/internal/kopia/data_collection_test.go

@@ -16,7 +16,9 @@ import (
 	"github.com/alcionai/corso/src/internal/common/readers"
 	"github.com/alcionai/corso/src/internal/data"
 	dataMock "github.com/alcionai/corso/src/internal/data/mock"
+	istats "github.com/alcionai/corso/src/internal/stats"
 	"github.com/alcionai/corso/src/internal/tester"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/path"
 )
@@ -395,12 +397,16 @@ func (suite *KopiaDataCollectionUnitSuite) TestFetchItemByName() {
 			defer flush()
 
 			root := getLayout(t, test.inputSerializationVersion)
-			c := &i64counter{}
+
+			counter := count.New()
+			c := istats.ByteCounter{
+				Counter: counter.AdderFor(count.PersistedUploadedBytes),
+			}
 
 			col := &kopiaDataCollection{
 				path:            pth,
 				dir:             root,
-				counter:         c,
+				counter:         &c,
 				expectedVersion: readers.DefaultSerializationVersion,
 			}
 

src/internal/kopia/inject/inject.go

@@ -8,6 +8,7 @@ import (
 	"github.com/alcionai/corso/src/internal/kopia"
 	"github.com/alcionai/corso/src/pkg/backup/details"
 	"github.com/alcionai/corso/src/pkg/backup/identity"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/path"
 )
@@ -23,6 +24,7 @@ type (
 			tags map[string]string,
 			buildTreeWithBase bool,
 			errs *fault.Bus,
+			counter *count.Bus,
 		) (*kopia.BackupStats, *details.Builder, kopia.DetailsMergeInfoer, error)
 	}
 

src/internal/kopia/merge_collection_test.go

@@ -15,7 +15,9 @@ import (
 	"github.com/alcionai/corso/src/internal/common/readers"
 	"github.com/alcionai/corso/src/internal/data"
 	"github.com/alcionai/corso/src/internal/m365/service/exchange/mock"
+	istats "github.com/alcionai/corso/src/internal/stats"
 	"github.com/alcionai/corso/src/internal/tester"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/path"
 )
@@ -271,15 +273,18 @@ func (suite *MergeCollectionUnitSuite) TestFetchItemByName() {
 			ctx, flush := tester.NewContext(t)
 			defer flush()
 
-			c := &i64counter{}
+			counter := count.New()
+			c := istats.ByteCounter{
+				Counter: counter.AdderFor(count.PersistedUploadedBytes),
+			}
 
 			dc := mergeCollection{fullPath: pth}
 
 			for i, layout := range layouts {
 				col := &kopiaDataCollection{
 					path:            pth,
 					dir:             layout(t),
-					counter:         c,
+					counter:         &c,
 					expectedVersion: readers.DefaultSerializationVersion,
 				}
 

src/internal/kopia/upload.go

@@ -59,6 +59,7 @@ type corsoProgress struct {
 	mu         sync.RWMutex
 	totalBytes int64
 	errs       *fault.Bus
+	counter    *count.Bus
 	// expectedIgnoredErrors is a count of error cases caught in the Error wrapper
 	// which are well known and actually ignorable.  At the end of a run, if the
 	// manifest ignored error count is equal to this count, then everything is good.
@@ -186,6 +187,7 @@ func (cp *corsoProgress) FinishedHashingFile(fname string, bs int64) {
 		"finished hashing file",
 		"path", clues.Hide(path.Elements(sl[2:])))
 
+	cp.counter.Add(count.PersistedHashedBytes, bs)
 	atomic.AddInt64(&cp.totalBytes, bs)
 }
 
@@ -214,7 +216,9 @@ func (cp *corsoProgress) Error(relpath string, err error, isIgnored bool) {
 	// delta query and a fetch.  This is our next point of error
 	// handling, where we can identify and skip over the case.
 	if clues.HasLabel(err, graph.LabelsSkippable) {
+		cp.counter.Inc(count.PersistenceExpectedErrors)
 		cp.incExpectedErrs()
+
 		return
 	}
 

src/internal/kopia/upload_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/alcionai/corso/src/internal/tester"
 	"github.com/alcionai/corso/src/pkg/backup/details"
 	"github.com/alcionai/corso/src/pkg/backup/identity"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/path"
 )
@@ -569,6 +570,7 @@ func (suite *CorsoProgressUnitSuite) TestFinishedHashingFile() {
 				deets:          bd,
 				pending:        map[string]*itemDetails{},
 				errs:           fault.New(true),
+				counter:        count.New(),
 			}
 
 			ci := test.cachedItems(suite.targetFileName, suite.targetFilePath)
@@ -579,6 +581,7 @@ func (suite *CorsoProgressUnitSuite) TestFinishedHashingFile() {
 
 			assert.Empty(t, cp.pending)
 			assert.Equal(t, test.expectedBytes, cp.totalBytes)
+			assert.Equal(t, test.expectedBytes, cp.counter.Get(count.PersistedHashedBytes))
 		})
 	}
 }
@@ -2669,7 +2672,7 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_SelectiveSubtreeP
 	getBaseSnapshot := func() (fs.Entry, map[string]*int) {
 		counters := map[string]*int{}
 
-		folder, count := newMockStaticDirectory(
+		folder, dirCount := newMockStaticDirectory(
 			encodeElements(folderID3)[0],
 			[]fs.Entry{
 				virtualfs.StreamingFileWithModTimeFromReader(
@@ -2681,9 +2684,9 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_SelectiveSubtreeP
 					time.Time{},
 					io.NopCloser(bytes.NewReader(fileData6))),
 			})
-		counters[folderID3] = count
+		counters[folderID3] = dirCount
 
-		folder, count = newMockStaticDirectory(
+		folder, dirCount = newMockStaticDirectory(
 			encodeElements(folderID2)[0],
 			[]fs.Entry{
 				virtualfs.StreamingFileWithModTimeFromReader(
@@ -2696,14 +2699,14 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_SelectiveSubtreeP
 					io.NopCloser(bytes.NewReader(fileData4))),
 				folder,
 			})
-		counters[folderID2] = count
+		counters[folderID2] = dirCount
 
-		folder4, count := newMockStaticDirectory(
+		folder4, dirCount := newMockStaticDirectory(
 			encodeElements(folderID4)[0],
 			[]fs.Entry{})
-		counters[folderID4] = count
+		counters[folderID4] = dirCount
 
-		folder, count = newMockStaticDirectory(
+		folder, dirCount = newMockStaticDirectory(
 			encodeElements(folderID1)[0],
 			[]fs.Entry{
 				virtualfs.StreamingFileWithModTimeFromReader(
@@ -2717,9 +2720,9 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_SelectiveSubtreeP
 				folder,
 				folder4,
 			})
-		counters[folderID1] = count
+		counters[folderID1] = dirCount
 
-		folder5, count := newMockStaticDirectory(
+		folder5, dirCount := newMockStaticDirectory(
 			encodeElements(folderID5)[0],
 			[]fs.Entry{
 				virtualfs.StreamingFileWithModTimeFromReader(
@@ -2731,7 +2734,7 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_SelectiveSubtreeP
 					time.Time{},
 					io.NopCloser(bytes.NewReader(fileData8))),
 			})
-		counters[folderID5] = count
+		counters[folderID5] = dirCount
 
 		return baseWithChildren(
 				prefixFolders,

src/internal/kopia/wrapper.go

@@ -26,6 +26,7 @@ import (
 	"github.com/alcionai/corso/src/pkg/backup/details"
 	"github.com/alcionai/corso/src/pkg/backup/identity"
 	"github.com/alcionai/corso/src/pkg/control/repository"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/logger"
 	"github.com/alcionai/corso/src/pkg/path"
@@ -76,6 +77,13 @@ func manifestToStats(
 	progress *corsoProgress,
 	uploadCount *stats.ByteCounter,
 ) BackupStats {
+	progress.counter.Add(count.PersistedFiles, int64(man.Stats.TotalFileCount))
+	progress.counter.Add(count.PersistedCachedFiles, int64(man.Stats.CachedFiles))
+	progress.counter.Add(count.PersistedNonCachedFiles, int64(man.Stats.NonCachedFiles))
+	progress.counter.Add(count.PersistedDirectories, int64(man.Stats.TotalDirectoryCount))
+	progress.counter.Add(count.PersistenceErrors, int64(man.Stats.ErrorCount))
+	progress.counter.Add(count.PersistenceIgnoredErrors, int64(man.Stats.IgnoredErrorCount))
+
 	return BackupStats{
 		SnapshotID: string(man.ID),
 
@@ -145,6 +153,7 @@ func (w Wrapper) ConsumeBackupCollections(
 	additionalTags map[string]string,
 	buildTreeWithBase bool,
 	errs *fault.Bus,
+	counter *count.Bus,
 ) (*BackupStats, *details.Builder, DetailsMergeInfoer, error) {
 	if w.c == nil {
 		return nil, nil, nil, clues.Stack(errNotConnected).WithClues(ctx)
@@ -163,6 +172,7 @@ func (w Wrapper) ConsumeBackupCollections(
 		deets:   &details.Builder{},
 		toMerge: newMergeDetails(),
 		errs:    errs,
+		counter: counter,
 	}
 
 	// When running an incremental backup, we need to pass the prior
@@ -227,7 +237,11 @@ func (w Wrapper) makeSnapshotWithRoot(
 ) (*BackupStats, error) {
 	var (
 		man *snapshot.Manifest
-		bc  = &stats.ByteCounter{}
+		bc  = &stats.ByteCounter{
+			// duplicate the count in the progress count.Bus.  Later we can
+			// replace the ByteCounter with the progress counter entirely.
+			Counter: progress.counter.AdderFor(count.PersistedUploadedBytes),
+		}
 	)
 
 	snapIDs := make([]manifest.ID, 0, len(prevSnapEntries)) // just for logging

src/internal/kopia/wrapper_scale_test.go

@@ -15,6 +15,7 @@ import (
 	exchMock "github.com/alcionai/corso/src/internal/m365/service/exchange/mock"
 	"github.com/alcionai/corso/src/internal/tester"
 	"github.com/alcionai/corso/src/pkg/backup/identity"
+	"github.com/alcionai/corso/src/pkg/count"
 	"github.com/alcionai/corso/src/pkg/fault"
 	"github.com/alcionai/corso/src/pkg/path"
 )
@@ -99,6 +100,7 @@ func BenchmarkHierarchyMerge(b *testing.B) {
 		base ManifestEntry,
 	) ManifestEntry {
 		bbs := test.baseBackups(base)
+		counter := count.New()
 
 		stats, _, _, err := w.ConsumeBackupCollections(
 			ctx,
@@ -108,11 +110,14 @@ func BenchmarkHierarchyMerge(b *testing.B) {
 			nil,
 			nil,
 			true,
-			fault.New(true))
+			fault.New(true),
+			counter)
 		require.NoError(t, err, clues.ToCore(err))
 
-		assert.Equal(t, 0, stats.IgnoredErrorCount)
-		assert.Equal(t, 0, stats.ErrorCount)
+		assert.Zero(t, stats.IgnoredErrorCount)
+		assert.Zero(t, stats.ErrorCount)
+		assert.Zero(t, counter.Get(count.PersistenceIgnoredErrors))
+		assert.Zero(t, counter.Get(count.PersistenceErrors))
 		assert.False(t, stats.Incomplete)
 
 		snap, err := snapshot.LoadSnapshot(