fix(mito2): keep SSTs when a manifest save ends with RateLimited or PermissionDenied (#9898)

* fix(mito2): keep SSTs when a manifest save ends with RateLimited

The object store's retry layer retries temporary errors and returns the
last attempt's error. A manifest delta PUT can be stored by the server
while the client sees a timeout, reset or 5xx, and the retries that
follow can be throttled, so the save ends with RateLimited even though
the delta exists. may_have_persisted_manifest_update treated RateLimited
as proof that nothing was written, so flush and compaction deleted the
SSTs the stored delta references. A later open or follower sync replays
the delta and reads fail on the missing files.

Treat RateLimited like other ambiguous errors: the SSTs are kept and
become orphans when the save really failed.

Fixes #9897

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* fix(mito2): keep SSTs when a manifest save ends with PermissionDenied

The retry layer returns the last attempt's error, so a 403 from a revoked
key, a changed bucket policy or clock skew can follow a delta PUT that
the server already stored. Like RateLimited, PermissionDenied does not
prove that nothing was written, so it no longer lets flush or compaction
delete the SSTs; they become orphans when the save really failed.

The compaction test that rejects a manifest write before persistence now
injects Unsupported, which is raised before any request is sent.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Co-authored-by: Yingwen <realevenyag@gmail.com>

src/mito2/src/engine/compaction_test.rs

@@ -1414,7 +1414,7 @@ impl object_store::layers::mock::Write for FailingManifestWriter {
     ) -> object_store::layers::mock::Result<()> {
         if matches!(self.failure, ManifestWriteFailure::BeforeWrite) {
             return Err(object_store::layers::mock::Error::new(
-                object_store::layers::mock::ErrorKind::PermissionDenied,
+                object_store::layers::mock::ErrorKind::Unsupported,
                 "manifest write rejected before persistence",
             ));
         }

src/mito2/src/error.rs

@@ -1445,7 +1445,9 @@ impl Error {
     /// Returns whether a failed manifest update may have been persisted.
     ///
     /// Unknown errors are treated conservatively because deleting output SSTs after an ambiguous
-    /// manifest write could leave the manifest referencing missing files.
+    /// manifest write could leave the manifest referencing missing files. The retry layer returns
+    /// the last attempt's error, so an error that can follow a stored attempt (throttling, or a 403
+    /// from a revoked key, a changed bucket policy or clock skew) never proves nothing was written.
     pub(crate) fn may_have_persisted_manifest_update(&self) -> bool {
         match self {
             Error::UpdateManifest { .. }
@@ -1459,11 +1461,9 @@ impl Error {
                 ErrorKind::Unsupported
                     | ErrorKind::ConfigInvalid
                     | ErrorKind::NotFound
-                    | ErrorKind::PermissionDenied
                     | ErrorKind::IsADirectory
                     | ErrorKind::NotADirectory
                     | ErrorKind::AlreadyExists
-                    | ErrorKind::RateLimited
                     | ErrorKind::IsSameFile
                     | ErrorKind::ConditionNotMatch
                     | ErrorKind::RangeNotSatisfied
@@ -1847,11 +1847,9 @@ mod tests {
             ErrorKind::Unsupported,
             ErrorKind::ConfigInvalid,
             ErrorKind::NotFound,
-            ErrorKind::PermissionDenied,
             ErrorKind::IsADirectory,
             ErrorKind::NotADirectory,
             ErrorKind::AlreadyExists,
-            ErrorKind::RateLimited,
             ErrorKind::IsSameFile,
             ErrorKind::ConditionNotMatch,
             ErrorKind::RangeNotSatisfied,
@@ -1864,9 +1862,18 @@ mod tests {
             );
         }
 
-        let error =
-            OpenDalSnafu {}.into_error(object_store::Error::new(ErrorKind::Unexpected, "test"));
-        assert!(error.may_have_persisted_manifest_update());
+        // The last retry can fail with these kinds even though an earlier attempt was stored.
+        for kind in [
+            ErrorKind::Unexpected,
+            ErrorKind::RateLimited,
+            ErrorKind::PermissionDenied,
+        ] {
+            let error = OpenDalSnafu {}.into_error(object_store::Error::new(kind, "test"));
+            assert!(
+                error.may_have_persisted_manifest_update(),
+                "error kind {kind:?} should not prove the manifest was not persisted"
+            );
+        }
 
         let region_id = RegionId::new(1, 1);
         let error = RegionTruncatedSnafu { region_id }.build();