log and continue if kopia can't find expected base snapshot directory (#2552)

## Description

This allows corso to continue making a snapshot even if the expected directory in the base is not found. The fact that the entry wasn't found will be logged

This should be safe so long as backups that had any sort of error, including "non-catastrophic" errors are marked as failed. Marking those as failed ensures that corso is using a matched set of metadata and data in the base snapshot rather than a partial snapshot

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

- [ ] ✅ Yes, it's included
- [x] 🕐 Yes, but in a later PR
- [ ] ⛔ No 

## Type of change

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

## Issue(s)

* #2550

## Test Plan

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

src/internal/kopia/upload.go

@@ -908,6 +908,11 @@ func inflateBaseTree(
 
 		ent, err := snapshotfs.GetNestedEntry(ctx, dir, pathElems)
 		if err != nil {
+			if isErrEntryNotFound(err) {
+				logger.Ctx(ctx).Infow("base snapshot missing subtree", "error", err)
+				continue
+			}
+
 			return errors.Wrapf(err, "snapshot %s getting subtree root", snap.ID)
 		}
 

src/internal/kopia/upload_test.go

@@ -2261,6 +2261,110 @@ func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTreeSkipsDeletedSubtre
 	expectTree(t, ctx, expected, dirTree)
 }
 
+func (suite *HierarchyBuilderUnitSuite) TestBuildDirectoryTree_HandleEmptyBase() {
+	tester.LogTimeOfTest(suite.T())
+	t := suite.T()
+
+	ctx, flush := tester.NewContext()
+	defer flush()
+
+	var (
+		archiveStorePath = makePath(
+			suite.T(),
+			[]string{testTenant, service, testUser, category, testArchiveID},
+			false)
+		archiveLocPath = makePath(
+			suite.T(),
+			[]string{testTenant, service, testUser, category, testArchiveDir},
+			false)
+	)
+
+	// baseSnapshot with the following layout:
+	// - a-tenant
+	//   - exchangeMetadata
+	//     - user1
+	//       - email
+	//         - file1
+	getBaseSnapshot := func() fs.Entry {
+		return baseWithChildren(
+			[]string{
+				testTenant,
+				path.ExchangeMetadataService.String(),
+				testUser,
+				category,
+			},
+			[]fs.Entry{
+				virtualfs.StreamingFileWithModTimeFromReader(
+					encodeElements(testFileName)[0],
+					time.Time{},
+					io.NopCloser(bytes.NewReader(testFileData)),
+				),
+			},
+		)
+	}
+
+	// Metadata subtree doesn't appear because we don't select it as one of the
+	// subpaths and we're not passing in a metadata collection.
+	expected := expectedTreeWithChildren(
+		[]string{
+			testTenant,
+			service,
+			testUser,
+			category,
+		},
+		[]*expectedNode{
+			{
+				name: testArchiveID,
+				children: []*expectedNode{
+					{
+						name:     testFileName2,
+						children: []*expectedNode{},
+					},
+				},
+			},
+		},
+	)
+
+	progress := &corsoProgress{
+		pending: map[string]*itemDetails{},
+		errs:    fault.New(true),
+	}
+	mc := mockconnector.NewMockExchangeCollection(archiveStorePath, archiveLocPath, 1)
+	mc.ColState = data.NewState
+	mc.Names[0] = testFileName2
+	mc.Data[0] = testFileData2
+
+	msw := &mockSnapshotWalker{
+		snapshotRoot: getBaseSnapshot(),
+	}
+
+	collections := []data.BackupCollection{mc}
+
+	// Returned directory structure should look like:
+	// - a-tenant
+	//   - exchangeMetadata
+	//     - user1
+	//       - emails
+	//         - file1
+	//   - exchange
+	//     - user1
+	//       - emails
+	//         - Archive
+	//           - file2
+	dirTree, err := inflateDirTree(
+		ctx,
+		msw,
+		[]IncrementalBase{
+			mockIncrementalBase("", testTenant, testUser, path.ExchangeService, path.EmailCategory),
+		},
+		collections,
+		nil,
+		progress)
+	require.NoError(t, err)
+
+	expectTree(t, ctx, expected, dirTree)
+}
+
 type mockMultiSnapshotWalker struct {
 	snaps map[string]fs.Entry
 }

src/internal/kopia/wrapper.go

@@ -331,7 +331,7 @@ func getItemStream(
 		encodeElements(itemPath.PopFront().Elements()...),
 	)
 	if err != nil {
-		if strings.Contains(err.Error(), "entry not found") {
+		if isErrEntryNotFound(err) {
 			err = clues.Stack(data.ErrNotFound, err).WithClues(ctx)
 		}
 
@@ -492,3 +492,8 @@ func (w Wrapper) FetchPrevSnapshotManifests(
 
 	return fetchPrevSnapshotManifests(ctx, w.c, reasons, tags), nil
 }
+
+func isErrEntryNotFound(err error) bool {
+	return strings.Contains(err.Error(), "entry not found") &&
+		!strings.Contains(err.Error(), "parent is not a directory")
+}

src/internal/tester/config.go

@@ -111,7 +111,7 @@ func readTestConfig() (map[string]string, error) {
 		TestCfgUserID,
 		os.Getenv(EnvCorsoM365TestUserID),
 		vpr.GetString(TestCfgUserID),
-		"conneri@8qzvrj.onmicrosoft.com",
+		"pradeepg@8qzvrj.onmicrosoft.com",
 	)
 	fallbackTo(
 		testEnv,
@@ -132,7 +132,7 @@ func readTestConfig() (map[string]string, error) {
 		TestCfgLoadTestOrgUsers,
 		os.Getenv(EnvCorsoM365LoadTestOrgUsers),
 		vpr.GetString(TestCfgLoadTestOrgUsers),
-		"lidiah@8qzvrj.onmicrosoft.com,conneri@8qzvrj.onmicrosoft.com",
+		"lidiah@8qzvrj.onmicrosoft.com,pradeepg@8qzvrj.onmicrosoft.com",
 	)
 	fallbackTo(
 		testEnv,