Skip exchange items which fail to download due to ErrorCorruptData error (#5191)

<!-- PR description-->

* We are seeing 500 errors from graph during exchange(email) backup. * Error is `:{"code":"ErrorCorruptData","message":"Data is corrupt., Invalid global object ID: some ID`. 
* Catch the error and add a skip since these items cannot be downloaded from graph. 

---

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

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

#### Type of change

<!--- Please check the type of change your PR introduces: --->
- [ ] 🌻 Feature
- [x] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Supportability/Tests
- [ ] 💻 CI/Deployment
- [ ] 🧹 Tech Debt/Cleanup

#### Issue(s)

<!-- Can reference multiple issues. Use one of the following "magic words" - "closes, fixes" to auto-close the Github issue. -->
* #<issue>

#### Test Plan

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

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased] (beta)
 
+### Fixed
+- Handle the case where an email or event cannot be retrieved from Exchange due to an `ErrorCorruptData` error. Corso will skip over the item but report it in the backup summary.
+
 ## [v0.19.0] (beta) - 2024-02-06
 
 ### Added

src/internal/m365/collection/exchange/collection.go

@@ -300,6 +300,19 @@ func (col *prefetchCollection) streamItems(
 						id,
 						map[string]any{"parentPath": parentPath}))
 					atomic.AddInt64(&success, 1)
+				case graph.IsErrCorruptData(err):
+					// These items cannot be downloaded, graph error indicates that the item
+					// data is corrupted. Add to skipped list.
+					logger.
+						CtxErr(ctx, err).
+						With("skipped_reason", fault.SkipCorruptData).
+						Info("inaccessible email")
+					errs.AddSkip(ctx, fault.EmailSkip(
+						fault.SkipCorruptData,
+						user,
+						id,
+						map[string]any{"parentPath": parentPath}))
+					atomic.AddInt64(&success, 1)
 				default:
 					col.Counter.Inc(count.StreamItemsErred)
 					el.AddRecoverable(ctx, clues.Wrap(err, "fetching item").Label(fault.LabelForceNoBackupCreation))

src/internal/m365/collection/exchange/collection_test.go

@@ -364,6 +364,17 @@ func (suite *CollectionUnitSuite) TestCollection_SkippedErrors() {
 			},
 			expectedSkipError: fault.EmailSkip(fault.SkipInvalidRecipients, "", "fisher", nil),
 		},
+		{
+			name: "ErrorCorruptData",
+			added: map[string]time.Time{
+				"fisher": {},
+			},
+			expectItemCount: 0,
+			itemGetter: &mock.ItemGetSerialize{
+				GetErr: graphTD.ODataErr(string(graph.ErrorCorruptData)),
+			},
+			expectedSkipError: fault.EmailSkip(fault.SkipCorruptData, "", "fisher", nil),
+		},
 	}
 
 	for _, test := range table {

src/pkg/fault/skipped.go

@@ -36,6 +36,10 @@ const (
 	// SkipInvalidRecipients identifies that an email was skipped because Exchange
 	// believes it is not valid and fails any attempt to read it.
 	SkipInvalidRecipients skipCause = "invalid_recipients_email"
+
+	// SkipCorruptData identifies that an email was skipped because graph reported
+	// that the email data was corrupt and failed all attempts to read it.
+	SkipCorruptData skipCause = "corrupt_data"
 )
 
 var _ print.Printable = &Skipped{}

src/pkg/services/m365/api/graph/errors.go

@@ -50,6 +50,10 @@ const (
 	// are mailbox creation racing with email receipt or a similar issue triggered
 	// due to on-prem->M365 mailbox migration.
 	ErrorInvalidRecipients errorCode = "ErrorInvalidRecipients"
+	// We have seen this graph error with a 500 response while backing up email
+	// messages. It implies that the email message is corrupt and cannot be read.
+	// Associated error message goes like "Data is corrupt.,Invalid global object ID:"
+	ErrorCorruptData errorCode = "ErrorCorruptData"
 	// This error occurs when an attempt is made to create a folder that has
 	// the same name as another folder in the same parent. Such duplicate folder
 	// names are not allowed by graph.
@@ -249,6 +253,10 @@ func IsErrInvalidRecipients(err error) bool {
 	return parseODataErr(err).hasErrorCode(err, ErrorInvalidRecipients)
 }
 
+func IsErrCorruptData(err error) bool {
+	return parseODataErr(err).hasErrorCode(err, ErrorCorruptData)
+}
+
 func IsErrCannotOpenFileAttachment(err error) bool {
 	return parseODataErr(err).hasErrorCode(err, cannotOpenFileAttachment)
 }

src/pkg/services/m365/api/graph/errors_test.go

@@ -235,7 +235,7 @@ func (suite *GraphErrorsUnitSuite) TestIsErrNotFound() {
 	}
 }
 
-func (suite *GraphErrorsUnitSuite) Test() {
+func (suite *GraphErrorsUnitSuite) TestIsErrInvalidRecipients() {
 	table := []struct {
 		name   string
 		err    error
@@ -269,6 +269,40 @@ func (suite *GraphErrorsUnitSuite) Test() {
 	}
 }
 
+func (suite *GraphErrorsUnitSuite) TestIsErrCorruptData() {
+	table := []struct {
+		name   string
+		err    error
+		expect assert.BoolAssertionFunc
+	}{
+		{
+			name:   "nil",
+			err:    nil,
+			expect: assert.False,
+		},
+		{
+			name:   "non-matching",
+			err:    assert.AnError,
+			expect: assert.False,
+		},
+		{
+			name:   "non-matching oDataErr",
+			err:    graphTD.ODataErr("fnords"),
+			expect: assert.False,
+		},
+		{
+			name:   "invalid receipient oDataErr",
+			err:    graphTD.ODataErr(string(ErrorCorruptData)),
+			expect: assert.True,
+		},
+	}
+	for _, test := range table {
+		suite.Run(test.name, func() {
+			test.expect(suite.T(), IsErrCorruptData(test.err))
+		})
+	}
+}
+
 func (suite *GraphErrorsUnitSuite) TestIsErrInvalidDelta() {
 	table := []struct {
 		name   string