[CORE-13370] archival: Fix retention fallthrough behavior

src/v/cloud_storage/recovery_errors.cc

@@ -14,11 +14,11 @@
 
 namespace cloud_storage {
 
-const char* recovery_error_category::name() const noexcept {
+const char* recovery_error_category_type::name() const noexcept {
     return "recovery_error_code";
 }
 
-std::string recovery_error_category::message(int c) const {
+std::string recovery_error_category_type::message(int c) const {
     switch (static_cast<recovery_error_code>(c)) {
     case recovery_error_code::success:
         return "recovery successful";
@@ -37,13 +37,13 @@ std::string recovery_error_category::message(int c) const {
     }
 }
 
-const std::error_category& error_category() noexcept {
-    static const recovery_error_category e;
+const std::error_category& recovery_error_category() noexcept {
+    static const recovery_error_category_type e;
     return e;
 }
 
 std::error_code make_error_code(recovery_error_code e) noexcept {
-    return {static_cast<int>(e), error_category()};
+    return {static_cast<int>(e), recovery_error_category()};
 }
 
 recovery_error_ctx

src/v/cloud_storage/recovery_errors.h

@@ -28,13 +28,13 @@ enum class recovery_error_code {
     unknown_error,
 };
 
-class recovery_error_category : public std::error_category {
+class recovery_error_category_type : public std::error_category {
 public:
     const char* name() const noexcept final;
     std::string message(int c) const final;
 };
 
-const std::error_category& error_category() noexcept;
+const std::error_category& recovery_error_category() noexcept;
 
 std::error_code make_error_code(recovery_error_code e) noexcept;
 

src/v/cluster/archival/ntp_archiver_service.cc

@@ -3018,6 +3018,23 @@ ss::future<> ntp_archiver::apply_spillover() {
     const auto manifest_upload_timeout = _conf->manifest_upload_timeout();
     const auto manifest_upload_backoff = _conf->cloud_storage_initial_backoff();
 
+    // Check the spillover invariant.
+    // The start_offset of the manifest must be equal to the begin_offset of
+    // the first segment in the manifest.
+    if (auto so = manifest().get_start_offset();
+        so.has_value() && !manifest().empty()) {
+        auto fo = manifest().begin()->base_offset;
+        if (fo != so.value()) {
+            vlog(
+              _rtclog.error,
+              "Spillover invariant violated: manifest start_offset {}, first "
+              "segment base_offset {}",
+              so.value(),
+              fo);
+            co_return;
+        }
+    }
+
     if (manifest_size_limit.has_value()) {
         vlog(
           _rtclog.debug,
@@ -3314,9 +3331,11 @@ ss::future<> ntp_archiver::apply_retention() {
         if (error != cluster::errc::success) {
             vlog(
               _rtclog.warn,
-              "Failed to update archival metadata STM start offest according "
+              "Failed to update archival metadata STM start offset according "
               "to retention policy: {}",
               error);
+            throw std::runtime_error(fmt_with_ctx(
+              fmt::format, "Failed to update start offset: {}", error));
         }
     } else {
         vlog(

tests/rptest/tests/cloud_storage_scrubber_test.py

@@ -54,6 +54,10 @@
     # manifest will refuse to apply the command and log the error below.
     r"cloud_storage - .* New replacement segment does not line up with previous segment",
     r"cluster - .* Can't add segment:",
+    # The same could happen with spillover. Spillover assumes that the start offset is
+    # aligned with the first segment in the manifest but this may not be the case after
+    # after the manifest reset.
+    r"archival - .* Spillover invariant violated",
 ]