Stricter checking for how hierarchies are merged (#1909)

## Description

Add some extra error checking for how the hierarchy can evolve during merging in kopia.Wrapper. Add more tests to solidify this behavior as well.

## 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
- [x] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [x] 🤖 Test
- [ ] 💻 CI/Deployment
- [ ] 🐹 Trivial/Minor

## Issue(s)

* closes #1884 

## Test Plan

<!-- How will this be tested prior to merging.-->
- [ ] 💪 Manual
- [x] ⚡ Unit test
- [ ] 💚 E2E

src/internal/kopia/upload.go

@@ -455,6 +455,29 @@ func newTreeMap() *treeMap {
 	}
 }
 
+// maybeGetTreeNode walks the tree(s) with roots roots and returns the node
+// specified by pathElements if all nodes on the path exist. If pathElements is
+// nil or empty then returns nil.
+func maybeGetTreeNode(roots map[string]*treeMap, pathElements []string) *treeMap {
+	if len(pathElements) == 0 {
+		return nil
+	}
+
+	dir := roots[pathElements[0]]
+
+	for i := 1; i < len(pathElements); i++ {
+		if dir == nil {
+			return nil
+		}
+
+		p := pathElements[i]
+
+		dir = dir.childDirs[p]
+	}
+
+	return dir
+}
+
 // getTreeNode walks the tree(s) with roots roots and returns the node specified
 // by pathElements. If pathElements is nil or empty then returns nil. Tree nodes
 // are created for any path elements where a node is not already present.
@@ -500,6 +523,9 @@ func inflateCollectionTree(
 	// Allows resolving what the new path should be when walking the base
 	// snapshot(s)'s hierarchy. Nil represents a collection that was deleted.
 	updatedPaths := make(map[string]path.Path)
+	// Temporary variable just to track the things that have been marked as
+	// changed while keeping a reference to their path.
+	changedPaths := []path.Path{}
 	ownerCats := &OwnersCats{
 		ResourceOwners: make(map[string]struct{}),
 		ServiceCats:    make(map[string]ServiceCat),
@@ -508,10 +534,29 @@ func inflateCollectionTree(
 	for _, s := range collections {
 		switch s.State() {
 		case data.DeletedState:
+			changedPaths = append(changedPaths, s.PreviousPath())
+
+			if _, ok := updatedPaths[s.PreviousPath().String()]; ok {
+				return nil, nil, errors.Errorf(
+					"multiple previous state changes to collection %s",
+					s.PreviousPath(),
+				)
+			}
+
 			updatedPaths[s.PreviousPath().String()] = nil
+
 			continue
 
 		case data.MovedState:
+			changedPaths = append(changedPaths, s.PreviousPath())
+
+			if _, ok := updatedPaths[s.PreviousPath().String()]; ok {
+				return nil, nil, errors.Errorf(
+					"multiple previous state changes to collection %s",
+					s.PreviousPath(),
+				)
+			}
+
 			updatedPaths[s.PreviousPath().String()] = s.FullPath()
 		}
 
@@ -531,9 +576,29 @@ func inflateCollectionTree(
 		ownerCats.ServiceCats[serviceCat] = ServiceCat{}
 		ownerCats.ResourceOwners[s.FullPath().ResourceOwner()] = struct{}{}
 
+		// Make sure there's only a single collection adding items for any given
+		// path in the new hierarchy.
+		if node.collection != nil {
+			return nil, nil, errors.Errorf("multiple instances of collection at %s", s.FullPath())
+		}
+
 		node.collection = s
 	}
 
+	// Check that each previous path has only one of the states of deleted, moved,
+	// or notmoved. Check at the end to avoid issues like seeing a notmoved state
+	// collection and then a deleted state collection.
+	for _, p := range changedPaths {
+		node := maybeGetTreeNode(roots, p.Elements())
+		if node == nil {
+			continue
+		}
+
+		if node.collection != nil && node.collection.State() == data.NotMovedState {
+			return nil, nil, errors.Errorf("conflicting states for collection %s", p)
+		}
+	}
+
 	return roots, updatedPaths, nil
 }
 

src/internal/kopia/upload_test.go

@@ -816,11 +816,93 @@ func mockIncrementalBase(
 	}
 }
 
+func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeErrors() {
+	dirPath := makePath(
+		suite.T(),
+		[]string{testTenant, service, testUser, category, testInboxDir},
+	)
+	dirPath2 := makePath(
+		suite.T(),
+		[]string{testTenant, service, testUser, category, testArchiveDir},
+	)
+
+	table := []struct {
+		name   string
+		states []data.CollectionState
+	}{
+		{
+			name: "DeletedAndNotMoved",
+			states: []data.CollectionState{
+				data.NotMovedState,
+				data.DeletedState,
+			},
+		},
+		{
+			name: "NotMovedAndDeleted",
+			states: []data.CollectionState{
+				data.DeletedState,
+				data.NotMovedState,
+			},
+		},
+		{
+			name: "DeletedAndMoved",
+			states: []data.CollectionState{
+				data.DeletedState,
+				data.MovedState,
+			},
+		},
+		{
+			name: "NotMovedAndMoved",
+			states: []data.CollectionState{
+				data.NotMovedState,
+				data.MovedState,
+			},
+		},
+	}
+
+	for _, test := range table {
+		suite.T().Run(test.name, func(t *testing.T) {
+			tester.LogTimeOfTest(t)
+
+			ctx, flush := tester.NewContext()
+			defer flush()
+
+			progress := &corsoProgress{pending: map[string]*itemDetails{}}
+
+			cols := []data.Collection{}
+			for _, s := range test.states {
+				prevPath := dirPath
+				nowPath := dirPath
+
+				switch s {
+				case data.DeletedState:
+					nowPath = nil
+				case data.MovedState:
+					nowPath = dirPath2
+				}
+
+				mc := mockconnector.NewMockExchangeCollection(nowPath, 0)
+				mc.ColState = s
+				mc.PrevPath = prevPath
+
+				cols = append(cols, mc)
+			}
+
+			_, err := inflateDirTree(ctx, nil, nil, cols, progress)
+			require.Error(t, err)
+		})
+	}
+}
+
 func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSingleSubtree() {
 	dirPath := makePath(
 		suite.T(),
 		[]string{testTenant, service, testUser, category, testInboxDir},
 	)
+	dirPath2 := makePath(
+		suite.T(),
+		[]string{testTenant, service, testUser, category, testArchiveDir},
+	)
 
 	// Must be a function that returns a new instance each time as StreamingFile
 	// can only return its Reader once.
@@ -942,6 +1024,85 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSingleSubtree() {
 				},
 			),
 		},
+		{
+			name: "DeleteAndNew",
+			inputCollections: func() []data.Collection {
+				mc1 := mockconnector.NewMockExchangeCollection(dirPath, 0)
+				mc1.ColState = data.DeletedState
+				mc1.PrevPath = dirPath
+
+				mc2 := mockconnector.NewMockExchangeCollection(dirPath, 1)
+				mc2.ColState = data.NewState
+				mc2.Names[0] = testFileName2
+				mc2.Data[0] = testFileData2
+
+				return []data.Collection{mc1, mc2}
+			},
+			expected: expectedTreeWithChildren(
+				[]string{
+					testTenant,
+					service,
+					testUser,
+					category,
+				},
+				[]*expectedNode{
+					{
+						name: testInboxDir,
+						children: []*expectedNode{
+							{
+								name:     testFileName2,
+								children: []*expectedNode{},
+								data:     testFileData2,
+							},
+						},
+					},
+				},
+			),
+		},
+		{
+			name: "MovedAndNew",
+			inputCollections: func() []data.Collection {
+				mc1 := mockconnector.NewMockExchangeCollection(dirPath2, 0)
+				mc1.ColState = data.MovedState
+				mc1.PrevPath = dirPath
+
+				mc2 := mockconnector.NewMockExchangeCollection(dirPath, 1)
+				mc2.ColState = data.NewState
+				mc2.Names[0] = testFileName2
+				mc2.Data[0] = testFileData2
+
+				return []data.Collection{mc1, mc2}
+			},
+			expected: expectedTreeWithChildren(
+				[]string{
+					testTenant,
+					service,
+					testUser,
+					category,
+				},
+				[]*expectedNode{
+					{
+						name: testInboxDir,
+						children: []*expectedNode{
+							{
+								name:     testFileName2,
+								children: []*expectedNode{},
+								data:     testFileData2,
+							},
+						},
+					},
+					{
+						name: testArchiveDir,
+						children: []*expectedNode{
+							{
+								name:     testFileName,
+								children: []*expectedNode{},
+							},
+						},
+					},
+				},
+			),
+		},
 	}
 
 	for _, test := range table {