Merge pull request #29622 from andrwng/ct-l1-lsm-exceptions

ct/l1/lsm: throw LSM exceptions from replicated_persistence

Author: andrwng

src/v/cloud_topics/level_one/metastore/lsm/replicated_persistence.cc

@@ -40,8 +40,17 @@ class replicated_metadata_persistence : public lsm::io::metadata_persistence {
         auto _ = _gate.hold();
         auto term_result = co_await _stm->sync(std::chrono::seconds(30), _as);
         if (!term_result.has_value()) {
-            throw lsm::io_error_exception(
-              "Failed to sync when loading manifest");
+            using enum stm::errc;
+            switch (term_result.error()) {
+            case shutting_down:
+                throw lsm::abort_requested_exception(
+                  "shutting down while syncing STM when loading manifest");
+            case not_leader:
+            case raft_error:
+                throw lsm::io_error_exception(
+                  "failed to sync STM when loading manifest: {}",
+                  term_result.error());
+            }
         }
         if (_stm->state().domain_uuid != _expected_domain_uuid) {
             // Caller needs to rebuild the metadata persistence with the
@@ -71,6 +80,7 @@ class replicated_metadata_persistence : public lsm::io::metadata_persistence {
 
     ss::future<>
     write_manifest(lsm::internal::database_epoch epoch, iobuf b) override {
+        _as.check();
         auto h = _gate.hold();
         co_await _cloud_persistence->write_manifest(epoch, b.share());
 
@@ -98,16 +108,25 @@ class replicated_metadata_persistence : public lsm::io::metadata_persistence {
 
         auto replicate_result = co_await _stm->replicate_and_wait(
           _stm->state().to_term(epoch), std::move(batch), _as);
-
         if (!replicate_result.has_value()) {
-            throw lsm::io_error_exception(
-              "Replication error after persisting manifest: {}",
-              int(replicate_result.error()));
+            using enum stm::errc;
+            switch (replicate_result.error()) {
+            case shutting_down:
+                throw lsm::abort_requested_exception(
+                  "shutting down while replicating manifest after persisting");
+            case not_leader:
+            case raft_error:
+                throw lsm::io_error_exception(
+                  "replication error after persisting manifest: {}",
+                  replicate_result.error());
+            }
         }
     }
 
     ss::future<> close() override {
-        _as.request_abort();
+        _as.request_abort_ex(
+          lsm::abort_requested_exception(
+            "replicated_persistence shutting down"));
         auto fut = _gate.close();
         auto persistence_fut = _cloud_persistence->close();
         co_await std::move(persistence_fut);

src/v/cloud_topics/level_one/metastore/lsm/stm.cc

@@ -45,6 +45,17 @@ void maybe_log_update_error(
       o,
       r.error());
 }
+
+std::string_view to_string_view(stm::errc e) {
+    switch (e) {
+    case stm::errc::not_leader:
+        return "stm::errc::not_leader";
+    case stm::errc::raft_error:
+        return "stm::errc::raft_error";
+    case stm::errc::shutting_down:
+        return "stm::errc::shutting_down";
+    }
+}
 } // namespace
 
 stm::stm(
@@ -258,4 +269,12 @@ void lsm_stm_factory::create(
     raft->log()->stm_manager()->add_stm(std::move(s));
 }
 
+std::ostream& operator<<(std::ostream& os, stm::errc e) {
+    return os << to_string_view(e);
+}
+
 } // namespace cloud_topics::l1
+
+auto format_as(cloud_topics::l1::stm::errc e) {
+    return cloud_topics::l1::to_string_view(e);
+}

src/v/cloud_topics/level_one/metastore/lsm/stm.h

@@ -100,4 +100,6 @@ class lsm_stm_factory : public cluster::state_machine_factory {
       const cluster::stm_instance_config&) final;
 };
 
+std::ostream& operator<<(std::ostream& os, stm::errc e);
+
 } // namespace cloud_topics::l1