introduce flat deletions to delta tree (#4719)

there are certain enumeration cases where tombstoning isn't the correct behavior, and we want to delete a folder from the tree entirely.  This primarily occurrs when we have a create->delete pair of markers either within or across enumerations.

---

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

- [x] ⛔ No

#### Type of change

- [ ] 🌻 Feature

#### Issue(s)

* #4689

#### Test Plan

- [x] ⚡ Unit test

src/internal/m365/collection/drive/collections_test.go

@@ -172,7 +172,7 @@ func malwareItem(
 }
 
 func driveRootItem(id string) models.DriveItemable {
-	name := "root"
+	name := rootName
 	item := models.NewDriveItem()
 	item.SetName(&name)
 	item.SetId(&id)
@@ -279,8 +279,8 @@ const (
 	malware   = "malware"
 	nav       = "nav"
 	pkg       = "package"
-	rootName  = "root"
+	rootID    = odConsts.RootID
-	rootID    = "root_id"
+	rootName  = odConsts.RootPathDir
 	subfolder = "subfolder"
 	tenant    = "t"
 	user      = "u"
@@ -1692,22 +1692,29 @@ func (suite *CollectionsUnitSuite) TestGet_treeCannotBeUsedWhileIncomplete() {
 	ctx, flush := tester.NewContext(t)
 	defer flush()
 
-	drive := models.NewDrive()
+	drv := models.NewDrive()
-	drive.SetId(ptr.To("id"))
+	drv.SetId(ptr.To("id"))
-	drive.SetName(ptr.To("name"))
+	drv.SetName(ptr.To("name"))
 
 	mbh := mock.DefaultOneDriveBH(user)
 	opts := control.DefaultOptions()
 	opts.ToggleFeatures.UseDeltaTree = true
 
-	mockDrivePager := &apiMock.Pager[models.Driveable]{
+	mbh.DrivePagerV = pagerForDrives(drv)
-		ToReturn: []apiMock.PagerResult[models.Driveable]{
+	mbh.DriveItemEnumeration = mock.EnumerateItemsDeltaByDrive{
-			{Values: []models.Driveable{drive}},
+		DrivePagers: map[string]*mock.DriveItemsDeltaPager{
+			"id": {
+				Pages: []mock.NextPage{{
+					Items: []models.DriveItemable{
+						driveRootItem(rootID), // will be present, not needed
+						delItem(id(file), parent(1), rootID, isFile),
+					},
+				}},
+				DeltaUpdate: pagers.DeltaUpdate{URL: id(delta)},
+			},
 		},
 	}
 
-	mbh.DrivePagerV = mockDrivePager
-
 	c := collWithMBH(mbh)
 	c.ctrl = opts
 

src/internal/m365/collection/drive/collections_tree.go

@@ -9,15 +9,21 @@ import (
 	"github.com/alcionai/corso/src/internal/common/prefixmatcher"
 	"github.com/alcionai/corso/src/internal/common/ptr"
 	"github.com/alcionai/corso/src/internal/data"
+	odConsts "github.com/alcionai/corso/src/internal/m365/service/onedrive/consts"
 	bupMD "github.com/alcionai/corso/src/pkg/backup/metadata"
 	"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"
 	"github.com/alcionai/corso/src/pkg/services/m365/api"
 	"github.com/alcionai/corso/src/pkg/services/m365/api/graph"
 	"github.com/alcionai/corso/src/pkg/services/m365/api/pagers"
 )
 
+// ---------------------------------------------------------------------------
+// Processing
+// ---------------------------------------------------------------------------
+
 // this file is used to separate the collections handling between the previous
 // (list-based) design, and the in-progress (tree-based) redesign.
 // see: https://github.com/alcionai/corso/issues/4688
@@ -29,6 +35,13 @@ func (c *Collections) getTree(
 	errs *fault.Bus,
 ) ([]data.BackupCollection, bool, error) {
 	ctx = clues.AddTraceName(ctx, "GetTree")
+	limiter := newPagerLimiter(c.ctrl)
+
+	logger.Ctx(ctx).Infow(
+		"running backup: getting collection data using tree structure",
+		"limits", c.ctrl.PreviewLimits,
+		"effective_limits", limiter.effectiveLimits(),
+		"preview_mode", limiter.enabled())
 
 	// extract the previous backup's metadata like: deltaToken urls and previousPath maps.
 	// We'll need these to reconstruct / ensure the correct state of the world, after
@@ -97,8 +110,10 @@ func (c *Collections) getTree(
 			ictx,
 			drv,
 			prevPathsByDriveID[driveID],
+			deltasByDriveID[driveID],
+			limiter,
 			cl,
-			el.Local())
+			el)
 		if err != nil {
 			el.AddRecoverable(ictx, clues.Stack(err))
 			continue
@@ -140,40 +155,52 @@ func (c *Collections) getTree(
 	return collections, canUsePrevBackup, nil
 }
 
+var errTreeNotImplemented = clues.New("backup tree not implemented")
+
 func (c *Collections) makeDriveCollections(
 	ctx context.Context,
-	d models.Driveable,
+	drv models.Driveable,
 	prevPaths map[string]string,
+	prevDeltaLink string,
+	limiter *pagerLimiter,
 	counter *count.Bus,
 	errs *fault.Bus,
 ) ([]data.BackupCollection, map[string]string, pagers.DeltaUpdate, error) {
-	cl := c.counter.Local()
+	ppfx, err := c.handler.PathPrefix(c.tenantID, ptr.Val(drv.GetId()))
+	if err != nil {
+		return nil, nil, pagers.DeltaUpdate{}, clues.Wrap(err, "generating backup tree prefix")
+	}
 
-	cl.Add(count.PrevPaths, int64(len(prevPaths)))
+	var (
-	logger.Ctx(ctx).Infow(
+		tree  = newFolderyMcFolderFace(ppfx)
-		"previous metadata for drive",
+		stats = &driveEnumerationStats{}
-		"count_old_prev_paths", len(prevPaths))
+	)
 
-	// TODO(keepers): leaving this code around for now as a guide
+	counter.Add(count.PrevPaths, int64(len(prevPaths)))
-	// while implementation progresses.
 
-	// --- pager aggregation
+	// --- delta item aggregation
 
-	// du, newPrevPaths, err := c.PopulateDriveCollections(
+	du, err := c.populateTree(
-	// 	ctx,
+		ctx,
-	// 	d,
+		tree,
-	// 	tree,
+		limiter,
-	// 	cl.Local(),
+		stats,
-	// 	errs)
+		drv,
-	// if err != nil {
+		prevDeltaLink,
-	// 	return nil, false, clues.Stack(err)
+		counter,
-	// }
+		errs)
+	if err != nil {
+		return nil, nil, pagers.DeltaUpdate{}, clues.Stack(err)
+	}
 
 	// numDriveItems := c.NumItems - numPrevItems
 	// numPrevItems = c.NumItems
 
 	// cl.Add(count.NewPrevPaths, int64(len(newPrevPaths)))
 
+	// TODO(keepers): leaving this code around for now as a guide
+	// while implementation progresses.
+
 	// --- prev path incorporation
 
 	// For both cases we don't need to do set difference on folder map if the
@@ -227,7 +254,285 @@ func (c *Collections) makeDriveCollections(
 	// 	}
 	// }
 
-	return nil, nil, pagers.DeltaUpdate{}, clues.New("not yet implemented")
+	// this is a dumb hack to satisfy the linter.
+	if ctx == nil {
+		return nil, nil, du, nil
+	}
+
+	return nil, nil, du, errTreeNotImplemented
+}
+
+// populateTree constructs a new tree and populates it with items
+// retrieved by enumerating the delta query for the drive.
+func (c *Collections) populateTree(
+	ctx context.Context,
+	tree *folderyMcFolderFace,
+	limiter *pagerLimiter,
+	stats *driveEnumerationStats,
+	drv models.Driveable,
+	prevDeltaLink string,
+	counter *count.Bus,
+	errs *fault.Bus,
+) (pagers.DeltaUpdate, error) {
+	ctx = clues.Add(ctx, "invalid_prev_delta", len(prevDeltaLink) == 0)
+
+	var (
+		driveID = ptr.Val(drv.GetId())
+		el      = errs.Local()
+	)
+
+	// TODO(keepers): to end in a correct state, we'll eventually need to run this
+	// query multiple times over, until it ends in an empty change set.
+	pager := c.handler.EnumerateDriveItemsDelta(
+		ctx,
+		driveID,
+		prevDeltaLink,
+		api.CallConfig{
+			Select: api.DefaultDriveItemProps(),
+		})
+
+	for page, reset, done := pager.NextPage(); !done; page, reset, done = pager.NextPage() {
+		if el.Failure() != nil {
+			break
+		}
+
+		counter.Inc(count.PagesEnumerated)
+
+		if reset {
+			counter.Inc(count.PagerResets)
+			tree.Reset()
+			c.resetStats()
+
+			*stats = driveEnumerationStats{}
+		}
+
+		err := c.enumeratePageOfItems(
+			ctx,
+			tree,
+			limiter,
+			stats,
+			drv,
+			page,
+			counter,
+			errs)
+		if err != nil {
+			el.AddRecoverable(ctx, clues.Stack(err))
+		}
+
+		// Stop enumeration early if we've reached the item or page limit. Do this
+		// at the end of the loop so we don't request another page in the
+		// background.
+		//
+		// We don't want to break on just the container limit here because it's
+		// possible that there's more items in the current (final) container that
+		// we're processing. We need to see the next page to determine if we've
+		// reached the end of the container. Note that this doesn't take into
+		// account the number of items in the current container, so it's possible it
+		// will fetch more data when it doesn't really need to.
+		if limiter.atPageLimit(stats) || limiter.atItemLimit(stats) {
+			break
+		}
+	}
+
+	// Always cancel the pager so that even if we exit early from the loop above
+	// we don't deadlock. Cancelling a pager that's already completed is
+	// essentially a noop.
+	pager.Cancel()
+
+	du, err := pager.Results()
+	if err != nil {
+		return du, clues.Stack(err)
+	}
+
+	logger.Ctx(ctx).Infow("enumerated collection delta", "stats", counter.Values())
+
+	return du, el.Failure()
+}
+
+func (c *Collections) enumeratePageOfItems(
+	ctx context.Context,
+	tree *folderyMcFolderFace,
+	limiter *pagerLimiter,
+	stats *driveEnumerationStats,
+	drv models.Driveable,
+	page []models.DriveItemable,
+	counter *count.Bus,
+	errs *fault.Bus,
+) error {
+	ctx = clues.Add(ctx, "page_lenth", len(page))
+	el := errs.Local()
+
+	for i, item := range page {
+		if el.Failure() != nil {
+			break
+		}
+
+		var (
+			isFolder = item.GetFolder() != nil || item.GetPackageEscaped() != nil
+			itemID   = ptr.Val(item.GetId())
+			err      error
+			skipped  *fault.Skipped
+		)
+
+		ictx := clues.Add(
+			ctx,
+			"item_id", itemID,
+			"item_name", clues.Hide(ptr.Val(item.GetName())),
+			"item_index", i,
+			"item_is_folder", isFolder,
+			"item_is_package", item.GetPackageEscaped() != nil)
+
+		if isFolder {
+			// check if the preview needs to exit before adding each folder
+			if !tree.ContainsFolder(itemID) && limiter.atLimit(stats, len(tree.folderIDToNode)) {
+				break
+			}
+
+			skipped, err = c.addFolderToTree(ictx, tree, drv, item, stats, counter)
+		} else {
+			skipped, err = c.addFileToTree(ictx, tree, drv, item, stats, counter)
+		}
+
+		if skipped != nil {
+			el.AddSkip(ctx, skipped)
+		}
+
+		if err != nil {
+			el.AddRecoverable(ictx, clues.Wrap(err, "adding folder"))
+		}
+
+		// Check if we reached the item or size limit while processing this page.
+		// The check after this loop will get us out of the pager.
+		// We don't want to check all limits because it's possible we've reached
+		// the container limit but haven't reached the item limit or really added
+		// items to the last container we found.
+		if limiter.atItemLimit(stats) {
+			break
+		}
+	}
+
+	stats.numPages++
+
+	return clues.Stack(el.Failure()).OrNil()
+}
+
+func (c *Collections) addFolderToTree(
+	ctx context.Context,
+	tree *folderyMcFolderFace,
+	drv models.Driveable,
+	folder models.DriveItemable,
+	stats *driveEnumerationStats,
+	counter *count.Bus,
+) (*fault.Skipped, error) {
+	var (
+		driveID     = ptr.Val(drv.GetId())
+		folderID    = ptr.Val(folder.GetId())
+		folderName  = ptr.Val(folder.GetName())
+		isDeleted   = folder.GetDeleted() != nil
+		isMalware   = folder.GetMalware() != nil
+		isPkg       = folder.GetPackageEscaped() != nil
+		parent      = folder.GetParentReference()
+		parentID    string
+		notSelected bool
+	)
+
+	if parent != nil {
+		parentID = ptr.Val(parent.GetId())
+	}
+
+	defer func() {
+		switch {
+		case notSelected:
+			counter.Inc(count.TotalContainersSkipped)
+		case isMalware:
+			counter.Inc(count.TotalMalwareProcessed)
+		case isDeleted:
+			counter.Inc(count.TotalDeleteFoldersProcessed)
+		case isPkg:
+			counter.Inc(count.TotalPackagesProcessed)
+		default:
+			counter.Inc(count.TotalFoldersProcessed)
+		}
+	}()
+
+	// FIXME(keepers): if we don't track this as previously visited,
+	// we could add a skip multiple times, every time we visit the
+	// folder again at the top of the page.
+	if isMalware {
+		skip := fault.ContainerSkip(
+			fault.SkipMalware,
+			driveID,
+			folderID,
+			folderName,
+			graph.ItemInfo(folder))
+
+		logger.Ctx(ctx).Infow("malware detected")
+
+		return skip, nil
+	}
+
+	if isDeleted {
+		err := tree.SetTombstone(ctx, folderID)
+		return nil, clues.Stack(err).OrNil()
+	}
+
+	collectionPath, err := c.makeFolderCollectionPath(ctx, driveID, folder)
+	if err != nil {
+		return nil, clues.Stack(err).Label(fault.LabelForceNoBackupCreation, count.BadCollPath)
+	}
+
+	// Skip items that don't match the folder selectors we were given.
+	notSelected = shouldSkip(ctx, collectionPath, c.handler, ptr.Val(drv.GetName()))
+	if notSelected {
+		logger.Ctx(ctx).Debugw("path not selected", "skipped_path", collectionPath.String())
+		return nil, nil
+	}
+
+	err = tree.SetFolder(ctx, parentID, folderID, folderName, isPkg)
+
+	return nil, clues.Stack(err).OrNil()
+}
+
+func (c *Collections) makeFolderCollectionPath(
+	ctx context.Context,
+	driveID string,
+	folder models.DriveItemable,
+) (path.Path, error) {
+	if folder.GetRoot() != nil {
+		pb := odConsts.DriveFolderPrefixBuilder(driveID)
+		collectionPath, err := c.handler.CanonicalPath(pb, c.tenantID)
+
+		return collectionPath, clues.WrapWC(ctx, err, "making canonical root path").OrNil()
+	}
+
+	if folder.GetParentReference() == nil || folder.GetParentReference().GetPath() == nil {
+		return nil, clues.NewWC(ctx, "no parent reference in folder").Label(count.MissingParent)
+	}
+
+	// Append folder name to path since we want the path for the collection, not
+	// the path for the parent of the collection.
+	name := ptr.Val(folder.GetName())
+	if len(name) == 0 {
+		return nil, clues.NewWC(ctx, "missing folder name")
+	}
+
+	folderPath := path.Split(ptr.Val(folder.GetParentReference().GetPath()))
+	folderPath = append(folderPath, name)
+	pb := path.Builder{}.Append(folderPath...)
+	collectionPath, err := c.handler.CanonicalPath(pb, c.tenantID)
+
+	return collectionPath, clues.WrapWC(ctx, err, "making folder collection path").OrNil()
+}
+
+func (c *Collections) addFileToTree(
+	ctx context.Context,
+	tree *folderyMcFolderFace,
+	drv models.Driveable,
+	item models.DriveItemable,
+	stats *driveEnumerationStats,
+	counter *count.Bus,
+) (*fault.Skipped, error) {
+	return nil, clues.New("not yet implemented")
 }
 
 // quality-of-life wrapper that transforms each tombstone in the map

src/internal/m365/collection/drive/delta_tree.go

@@ -36,6 +36,9 @@ type folderyMcFolderFace struct {
 	// it's just a sensible place to store the data, since we're
 	// already pushing file additions through the api.
 	excludeFileIDs map[string]struct{}
+
+	// true if Reset() was called
+	hadReset bool
 }
 
 func newFolderyMcFolderFace(
@@ -49,6 +52,18 @@ func newFolderyMcFolderFace(
 	}
 }
 
+// Reset erases all data contained in the tree.  This is intended for
+// tracking a delta enumeration reset, not for tree re-use, and will
+// cause the tree to flag itself as dirty in order to appropriately
+// post-process the data.
+func (face *folderyMcFolderFace) Reset() {
+	face.hadReset = true
+	face.root = nil
+	face.folderIDToNode = map[string]*nodeyMcNodeFace{}
+	face.tombstones = map[string]*nodeyMcNodeFace{}
+	face.excludeFileIDs = map[string]struct{}{}
+}
+
 type nodeyMcNodeFace struct {
 	// required for mid-enumeration folder moves, else we have to walk
 	// the tree completely to remove the node from its old parent.
@@ -71,14 +86,12 @@ type nodeyMcNodeFace struct {
 func newNodeyMcNodeFace(
 	parent *nodeyMcNodeFace,
 	id, name string,
-	prev path.Elements,
 	isPackage bool,
 ) *nodeyMcNodeFace {
 	return &nodeyMcNodeFace{
 		parent:    parent,
 		id:        id,
 		name:      name,
-		prev:      prev,
 		children:  map[string]*nodeyMcNodeFace{},
 		items:     map[string]time.Time{},
 		isPackage: isPackage,
@@ -89,6 +102,21 @@ func newNodeyMcNodeFace(
 // folder handling
 // ---------------------------------------------------------------------------
 
+// ContainsFolder returns true if the given folder id is present as either
+// a live node or a tombstone.
+func (face *folderyMcFolderFace) ContainsFolder(id string) bool {
+	_, stillKicking := face.folderIDToNode[id]
+	_, alreadyBuried := face.tombstones[id]
+
+	return stillKicking || alreadyBuried
+}
+
+// CountNodes returns a count that is the sum of live folders and
+// tombstones recorded in the tree.
+func (face *folderyMcFolderFace) CountFolders() int {
+	return len(face.tombstones) + len(face.folderIDToNode)
+}
+
 // SetFolder adds a node with the following details to the tree.
 // If the node already exists with the given ID, the name and parent
 // values are updated to match (isPackage is assumed not to change).
@@ -118,7 +146,7 @@ func (face *folderyMcFolderFace) SetFolder(
 	// only set the root node once.
 	if name == odConsts.RootPathDir {
 		if face.root == nil {
-			root := newNodeyMcNodeFace(nil, id, name, nil, isPackage)
+			root := newNodeyMcNodeFace(nil, id, name, isPackage)
 			face.root = root
 			face.folderIDToNode[id] = root
 		}
@@ -178,9 +206,7 @@ func (face *folderyMcFolderFace) SetFolder(
 		nodey.parent = parent
 	} else {
 		// change type 1: new addition
-		// the previous location is always nil, since previous path additions get their
+		nodey = newNodeyMcNodeFace(parent, id, name, isPackage)
-		// own setter func.
-		nodey = newNodeyMcNodeFace(parent, id, name, nil, isPackage)
 	}
 
 	// ensure the parent points to this node, and that the node is registered
@@ -194,16 +220,11 @@ func (face *folderyMcFolderFace) SetFolder(
 func (face *folderyMcFolderFace) SetTombstone(
 	ctx context.Context,
 	id string,
-	loc path.Elements,
 ) error {
 	if len(id) == 0 {
 		return clues.NewWC(ctx, "missing tombstone folder ID")
 	}
 
-	if len(loc) == 0 {
-		return clues.NewWC(ctx, "missing tombstone location")
-	}
-
 	// since we run mutiple enumerations, it's possible to see a folder added on the
 	// first enumeration that then gets deleted on the next.  This means that the folder
 	// was deleted while the first enumeration was in flight, and will show up even if
@@ -225,29 +246,8 @@ func (face *folderyMcFolderFace) SetTombstone(
 		return nil
 	}
 
-	zombey, alreadyBuried := face.tombstones[id]
+	if _, alreadyBuried := face.tombstones[id]; !alreadyBuried {
-	if alreadyBuried {
+		face.tombstones[id] = newNodeyMcNodeFace(nil, id, "", false)
-		if zombey.prev.String() != loc.String() {
-			// logging for sanity
-			logger.Ctx(ctx).Infow(
-				"attempted to tombstone two paths with the same ID",
-				"first_tombstone_path", zombey.prev,
-				"second_tombstone_path", loc)
-		}
-
-		// since we're storing drive data by folder name in kopia, not id, we need
-		// to make sure to preserve the original tombstone location.  If we get a
-		// conflicting set of locations in the same delta enumeration, we can always
-		// treat the original one as the canonical one.  IE: what we're deleting is
-		// the original location as it exists in kopia.  So even if we get a newer
-		// location in the drive enumeration, the original location is the one that
-		// kopia uses, and the one we need to tombstone.
-		//
-		// this should also be asserted in the second step, where we compare the delta
-		// changes to the backup previous paths metadata.
-		face.tombstones[id] = zombey
-	} else {
-		face.tombstones[id] = newNodeyMcNodeFace(nil, id, "", loc, false)
 	}
 
 	return nil

src/internal/m365/collection/drive/delta_tree_test.go

@@ -8,11 +8,49 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
-	odConsts "github.com/alcionai/corso/src/internal/m365/service/onedrive/consts"
 	"github.com/alcionai/corso/src/internal/tester"
 	"github.com/alcionai/corso/src/pkg/path"
 )
 
+// ---------------------------------------------------------------------------
+// helpers
+// ---------------------------------------------------------------------------
+
+var loc = path.NewElements("root:/foo/bar/baz/qux/fnords/smarf/voi/zumba/bangles/howdyhowdyhowdy")
+
+func treeWithRoot() *folderyMcFolderFace {
+	tree := newFolderyMcFolderFace(nil)
+	rootey := newNodeyMcNodeFace(nil, rootID, rootName, false)
+	tree.root = rootey
+	tree.folderIDToNode[rootID] = rootey
+
+	return tree
+}
+
+func treeWithTombstone() *folderyMcFolderFace {
+	tree := treeWithRoot()
+	tree.tombstones[id(folder)] = newNodeyMcNodeFace(nil, id(folder), "", false)
+
+	return tree
+}
+
+func treeWithFolders() *folderyMcFolderFace {
+	tree := treeWithRoot()
+
+	o := newNodeyMcNodeFace(tree.root, idx(folder, "parent"), namex(folder, "parent"), true)
+	tree.folderIDToNode[o.id] = o
+
+	f := newNodeyMcNodeFace(o, id(folder), name(folder), false)
+	tree.folderIDToNode[f.id] = f
+	o.children[f.id] = f
+
+	return tree
+}
+
+// ---------------------------------------------------------------------------
+// tests
+// ---------------------------------------------------------------------------
+
 type DeltaTreeUnitSuite struct {
 	tester.Suite
 }
@@ -24,7 +62,7 @@ func TestDeltaTreeUnitSuite(t *testing.T) {
 func (suite *DeltaTreeUnitSuite) TestNewFolderyMcFolderFace() {
 	var (
 		t      = suite.T()
-		p, err = path.BuildPrefix("t", "r", path.OneDriveService, path.FilesCategory)
+		p, err = path.BuildPrefix(tenant, user, path.OneDriveService, path.FilesCategory)
 	)
 
 	require.NoError(t, err, clues.ToCore(err))
@@ -41,14 +79,13 @@ func (suite *DeltaTreeUnitSuite) TestNewNodeyMcNodeFace() {
 	var (
 		t      = suite.T()
 		parent = &nodeyMcNodeFace{}
-		loc    = path.NewElements("root:/foo/bar/baz/qux/fnords/smarf/voi/zumba/bangles/howdyhowdyhowdy")
 	)
 
-	nodeFace := newNodeyMcNodeFace(parent, "id", "name", loc, true)
+	nodeFace := newNodeyMcNodeFace(parent, "id", "name", true)
 	assert.Equal(t, parent, nodeFace.parent)
 	assert.Equal(t, "id", nodeFace.id)
 	assert.Equal(t, "name", nodeFace.name)
-	assert.Equal(t, loc, nodeFace.prev)
+	assert.NotEqual(t, loc, nodeFace.prev)
 	assert.True(t, nodeFace.isPackage)
 	assert.NotNil(t, nodeFace.children)
 	assert.NotNil(t, nodeFace.items)
@@ -57,35 +94,6 @@ func (suite *DeltaTreeUnitSuite) TestNewNodeyMcNodeFace() {
 // note that this test is focused on the SetFolder function,
 // and intentionally does not verify the resulting node tree
 func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder() {
-	treeWithRoot := func() *folderyMcFolderFace {
-		rootey := newNodeyMcNodeFace(nil, odConsts.RootID, odConsts.RootPathDir, nil, false)
-		tree := newFolderyMcFolderFace(nil)
-		tree.root = rootey
-		tree.folderIDToNode[odConsts.RootID] = rootey
-
-		return tree
-	}
-
-	treeWithTombstone := func() *folderyMcFolderFace {
-		tree := treeWithRoot()
-		tree.tombstones["folder"] = newNodeyMcNodeFace(nil, "folder", "", path.NewElements(""), false)
-
-		return tree
-	}
-
-	treeWithFolders := func() *folderyMcFolderFace {
-		tree := treeWithRoot()
-
-		o := newNodeyMcNodeFace(tree.root, "other", "o", nil, true)
-		tree.folderIDToNode[o.id] = o
-
-		f := newNodeyMcNodeFace(o, "folder", "f", nil, false)
-		tree.folderIDToNode[f.id] = f
-		o.children[f.id] = f
-
-		return tree
-	}
-
 	table := []struct {
 		tname     string
 		tree      *folderyMcFolderFace
@@ -94,102 +102,91 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder() {
 		name      string
 		isPackage bool
 		expectErr assert.ErrorAssertionFunc
-		expectPrev assert.ValueAssertionFunc
 	}{
 		{
 			tname: "add root",
 			tree: &folderyMcFolderFace{
 				folderIDToNode: map[string]*nodeyMcNodeFace{},
 			},
-			id:         odConsts.RootID,
+			id:        rootID,
-			name:       odConsts.RootPathDir,
+			name:      rootName,
 			isPackage: true,
 			expectErr: assert.NoError,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "root already exists",
 			tree:      treeWithRoot(),
-			id:         odConsts.RootID,
+			id:        rootID,
-			name:       odConsts.RootPathDir,
+			name:      rootName,
 			expectErr: assert.NoError,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "add folder",
 			tree:      treeWithRoot(),
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			expectErr: assert.NoError,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "add package",
 			tree:      treeWithRoot(),
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			isPackage: true,
 			expectErr: assert.NoError,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "missing ID",
 			tree:      treeWithRoot(),
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			isPackage: true,
 			expectErr: assert.Error,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "missing name",
 			tree:      treeWithRoot(),
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			id:         "folder",
+			id:        id(folder),
 			isPackage: true,
 			expectErr: assert.Error,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "missing parentID",
 			tree:      treeWithRoot(),
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			isPackage: true,
 			expectErr: assert.Error,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "already tombstoned",
 			tree:      treeWithTombstone(),
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			expectErr: assert.NoError,
-			expectPrev: assert.NotNil,
 		},
 		{
 			tname: "add folder before parent",
 			tree: &folderyMcFolderFace{
 				folderIDToNode: map[string]*nodeyMcNodeFace{},
 			},
-			parentID:   odConsts.RootID,
+			parentID:  rootID,
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			isPackage: true,
 			expectErr: assert.Error,
-			expectPrev: assert.Nil,
 		},
 		{
 			tname:     "folder already exists",
 			tree:      treeWithFolders(),
-			parentID:   "other",
+			parentID:  idx(folder, "parent"),
-			id:         "folder",
+			id:        id(folder),
-			name:       "nameyMcNameFace",
+			name:      name(folder),
 			expectErr: assert.NoError,
-			expectPrev: assert.Nil,
 		},
 	}
 	for _, test := range table {
@@ -213,7 +210,6 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder() {
 
 			result := test.tree.folderIDToNode[test.id]
 			require.NotNil(t, result)
-			test.expectPrev(t, result.prev)
 			assert.Equal(t, test.id, result.id)
 			assert.Equal(t, test.name, result.name)
 			assert.Equal(t, test.isPackage, result.isPackage)
@@ -230,65 +226,38 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder() {
 }
 
 func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_AddTombstone() {
-	loc := path.NewElements("root:/foo/bar/baz/qux/fnords/smarf/voi/zumba/bangles/howdyhowdyhowdy")
-	treeWithTombstone := func() *folderyMcFolderFace {
-		tree := newFolderyMcFolderFace(nil)
-		tree.tombstones["id"] = newNodeyMcNodeFace(nil, "id", "", loc, false)
-
-		return tree
-	}
-
 	table := []struct {
 		name      string
 		id        string
-		loc       path.Elements
 		tree      *folderyMcFolderFace
 		expectErr assert.ErrorAssertionFunc
 	}{
 		{
 			name:      "add tombstone",
-			id:        "id",
+			id:        id(folder),
-			loc:       loc,
 			tree:      newFolderyMcFolderFace(nil),
 			expectErr: assert.NoError,
 		},
 		{
 			name:      "duplicate tombstone",
-			id:        "id",
+			id:        id(folder),
-			loc:       loc,
 			tree:      treeWithTombstone(),
 			expectErr: assert.NoError,
 		},
 		{
 			name:      "missing ID",
-			loc:       loc,
-			tree:      newFolderyMcFolderFace(nil),
-			expectErr: assert.Error,
-		},
-		{
-			name:      "missing loc",
-			id:        "id",
-			tree:      newFolderyMcFolderFace(nil),
-			expectErr: assert.Error,
-		},
-		{
-			name:      "empty loc",
-			id:        "id",
-			loc:       path.Elements{},
 			tree:      newFolderyMcFolderFace(nil),
 			expectErr: assert.Error,
 		},
 		{
 			name:      "conflict: folder alive",
-			id:        "id",
+			id:        id(folder),
-			loc:       loc,
 			tree:      treeWithTombstone(),
 			expectErr: assert.NoError,
 		},
 		{
-			name:      "already tombstoned with different path",
+			name:      "already tombstoned",
-			id:        "id",
+			id:        id(folder),
-			loc:       append(path.Elements{"foo"}, loc...),
 			tree:      treeWithTombstone(),
 			expectErr: assert.NoError,
 		},
@@ -300,7 +269,7 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_AddTombstone() {
 			ctx, flush := tester.NewContext(t)
 			defer flush()
 
-			err := test.tree.SetTombstone(ctx, test.id, test.loc)
+			err := test.tree.SetTombstone(ctx, test.id)
 			test.expectErr(t, err, clues.ToCore(err))
 
 			if err != nil {
@@ -309,8 +278,6 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_AddTombstone() {
 
 			result := test.tree.tombstones[test.id]
 			require.NotNil(t, result)
-			require.NotEmpty(t, result.prev)
-			assert.Equal(t, loc, result.prev)
 		})
 	}
 }
@@ -431,22 +398,17 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder_correctTree()
 	ctx, flush := tester.NewContext(t)
 	defer flush()
 
-	tree := newFolderyMcFolderFace(nil)
+	tree := treeWithRoot()
-	rootID := odConsts.RootID
 
 	set := func(
-		parentID, id, name string,
+		parentID, fid, fname string,
 		isPackage bool,
 	) {
-		err := tree.SetFolder(ctx, parentID, id, name, isPackage)
+		err := tree.SetFolder(ctx, parentID, fid, fname, isPackage)
 		require.NoError(t, err, clues.ToCore(err))
 	}
 
-	assert.Nil(t, tree.root)
+	// assert the root exists
-	assert.Empty(t, tree.folderIDToNode)
-
-	// add the root
-	set("", rootID, odConsts.RootPathDir, false)
 
 	assert.NotNil(t, tree.root)
 	assert.Equal(t, rootID, tree.root.id)
@@ -455,18 +417,18 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder_correctTree()
 	an(tree.root).compare(t, tree, true)
 
 	// add a child at the root
-	set(tree.root.id, "lefty", "l", false)
+	set(rootID, id("lefty"), name("l"), false)
 
-	lefty := tree.folderIDToNode["lefty"]
+	lefty := tree.folderIDToNode[id("lefty")]
 	an(
 		tree.root,
 		an(lefty)).
 		compare(t, tree, true)
 
 	// add another child at the root
-	set(tree.root.id, "righty", "r", false)
+	set(rootID, id("righty"), name("r"), false)
 
-	righty := tree.folderIDToNode["righty"]
+	righty := tree.folderIDToNode[id("righty")]
 	an(
 		tree.root,
 		an(lefty),
@@ -474,9 +436,9 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder_correctTree()
 		compare(t, tree, true)
 
 	// add a child to lefty
-	set(lefty.id, "bloaty", "bl", false)
+	set(lefty.id, id("bloaty"), name("bl"), false)
 
-	bloaty := tree.folderIDToNode["bloaty"]
+	bloaty := tree.folderIDToNode[id("bloaty")]
 	an(
 		tree.root,
 		an(lefty, an(bloaty)),
@@ -484,9 +446,9 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder_correctTree()
 		compare(t, tree, true)
 
 	// add another child to lefty
-	set(lefty.id, "brightly", "br", false)
+	set(lefty.id, id("brightly"), name("br"), false)
 
-	brightly := tree.folderIDToNode["brightly"]
+	brightly := tree.folderIDToNode[id("brightly")]
 	an(
 		tree.root,
 		an(lefty, an(bloaty), an(brightly)),
@@ -522,40 +484,34 @@ func (suite *DeltaTreeUnitSuite) TestFolderyMcFolderFace_SetFolder_correctTombst
 	ctx, flush := tester.NewContext(t)
 	defer flush()
 
-	tree := newFolderyMcFolderFace(nil)
+	tree := treeWithRoot()
-	rootID := odConsts.RootID
 
 	set := func(
-		parentID, id, name string,
+		parentID, fid, fname string,
 		isPackage bool,
 	) {
-		err := tree.SetFolder(ctx, parentID, id, name, isPackage)
+		err := tree.SetFolder(ctx, parentID, fid, fname, isPackage)
 		require.NoError(t, err, clues.ToCore(err))
 	}
 
 	tomb := func(
-		id string,
+		tid string,
 		loc path.Elements,
 	) {
-		err := tree.SetTombstone(ctx, id, loc)
+		err := tree.SetTombstone(ctx, tid)
 		require.NoError(t, err, clues.ToCore(err))
 	}
 
-	assert.Nil(t, tree.root)
-	assert.Empty(t, tree.folderIDToNode)
-
 	// create a simple tree
 	// root > branchy > [leafy, bob]
-	set("", rootID, odConsts.RootPathDir, false)
+	set(tree.root.id, id("branchy"), name("br"), false)
+	branchy := tree.folderIDToNode[id("branchy")]
 
-	set(tree.root.id, "branchy", "br", false)
+	set(branchy.id, id("leafy"), name("l"), false)
-	branchy := tree.folderIDToNode["branchy"]
+	set(branchy.id, id("bob"), name("bobbers"), false)
 
-	set(branchy.id, "leafy", "l", false)
+	leafy := tree.folderIDToNode[id("leafy")]
-	set(branchy.id, "bob", "bobbers", false)
+	bob := tree.folderIDToNode[id("bob")]
-
-	leafy := tree.folderIDToNode["leafy"]
-	bob := tree.folderIDToNode["bob"]
 
 	an(
 		tree.root,

src/internal/m365/collection/drive/limiter_test.go

@@ -26,6 +26,26 @@ import (
 	"github.com/alcionai/corso/src/pkg/services/m365/api/pagers"
 )
 
+// ---------------------------------------------------------------------------
+// helpers
+// ---------------------------------------------------------------------------
+
+func minimumLimitOpts() control.Options {
+	minLimitOpts := control.DefaultOptions()
+	minLimitOpts.PreviewLimits.Enabled = true
+	minLimitOpts.PreviewLimits.MaxBytes = 1
+	minLimitOpts.PreviewLimits.MaxContainers = 1
+	minLimitOpts.PreviewLimits.MaxItems = 1
+	minLimitOpts.PreviewLimits.MaxItemsPerContainer = 1
+	minLimitOpts.PreviewLimits.MaxPages = 1
+
+	return minLimitOpts
+}
+
+// ---------------------------------------------------------------------------
+// tests
+// ---------------------------------------------------------------------------
+
 type LimiterUnitSuite struct {
 	tester.Suite
 }

src/pkg/count/keys.go

@@ -9,10 +9,8 @@ const (
 	ThrottledAPICalls Key = "throttled-api-calls"
 )
 
-// Tracked during backup
+// backup amounts reported by kopia
 const (
-	// amounts reported by kopia
-
 	PersistedCachedFiles          Key = "persisted-cached-files"
 	PersistedDirectories          Key = "persisted-directories"
 	PersistedFiles                Key = "persisted-files"
@@ -24,9 +22,10 @@ const (
 	PersistenceErrors             Key = "persistence-errors"
 	PersistenceExpectedErrors     Key = "persistence-expected-errors"
 	PersistenceIgnoredErrors      Key = "persistence-ignored-errors"
+)
 
-	// amounts reported by data providers
+// backup amounts reported by data providers
-
+const (
 	Channels                      Key = "channels"
 	CollectionMoved               Key = "collection-moved"
 	CollectionNew                 Key = "collection-state-new"
@@ -66,10 +65,27 @@ const (
 	StreamItemsErrored            Key = "stream-items-errored"
 	StreamItemsFound              Key = "stream-items-found"
 	StreamItemsRemoved            Key = "stream-items-removed"
+	TotalContainersSkipped        Key = "total-containers-skipped"
 	URLCacheMiss                  Key = "url-cache-miss"
 	URLCacheRefresh               Key = "url-cache-refresh"
+)
 
-	// miscellaneous
+// Total___Processed counts are used to track raw processing numbers
+// for values that may have a similar, but different, end result count.
+// For example: a delta query may add the same folder to many different pages.
+// instead of adding logic to catch folder duplications and only count new
+// entries, we can increment TotalFoldersProcessed for every duplication,
+// and use a separate Key (Folders) for the end count of folders produced
+// at the end of the delta enumeration.
+const (
+	TotalDeleteFoldersProcessed Key = "total-delete-folders-processed"
+	TotalFoldersProcessed       Key = "total-folders-processed"
+	TotalMalwareProcessed       Key = "total-malware-processed"
+	TotalPackagesProcessed      Key = "total-packages-processed"
+)
+
+// miscellaneous
+const (
 	RequiresUserPnToIDMigration Key = "requires-user-pn-to-id-migration"
 )
 

src/pkg/logger/logger.go

@@ -491,6 +491,7 @@ func CtxStack(ctx context.Context, skip int) *zap.SugaredLogger {
 // CtxErr retrieves the logger embedded in the context
 // and packs all of the structured data in the error inside it.
 func CtxErr(ctx context.Context, err error) *zap.SugaredLogger {
+	// TODO(keepers): only log the err values, not the ctx.
 	return Ctx(ctx).
 		With(
 			"error", err,