replica: add abort_requested_exception to exception_variant

If migration_manager::get_schema_for_write is called after
migration_manager::drain, it throws abort_requested_exception.
This exception is not present in replica::exception_variant, which
means that RPC doesn't preserve information about its type. If it is
thrown on the replica side, it is deserialized as std::runtime_error
on the coordinator. Therefore, abstract_read_resolver::error logs
information about this exception, even though we don't want it (aborts
are triggered on shutdown and timeouts).

To solve this issue, we add abort_requested_exception to
replica::exception_variant and, in the next commits, refactor
storage_proxy::handle_read so that abort_requested_exception thrown in
migration_manager::get_schema_for_write is properly serialized. Thanks
to this change, unchanged abstract_read_resolver::error correctly
handles abort_requested_exception thrown on the replica side by not
reporting it.

idl/replica_exception.idl.hh

@@ -20,11 +20,15 @@ class stale_topology_exception {
     int64_t callee_fence_version();
 };
 
+class abort_requested_exception {
+};
+
 struct exception_variant {
     std::variant<replica::unknown_exception,
             replica::no_exception,
             replica::rate_limit_exception,
-            replica::stale_topology_exception
+            replica::stale_topology_exception,
+            replica::abort_requested_exception
     > reason;
 };
 

replica/exceptions.cc

@@ -24,6 +24,8 @@ exception_variant try_encode_replica_exception(std::exception_ptr eptr) {
         return rate_limit_exception();
     } catch (const stale_topology_exception& e) {
         return e;
+    } catch (abort_requested_exception&) {
+        return abort_requested_exception();
     } catch (...) {
         return no_exception{};
     }

replica/exceptions.hh

@@ -13,6 +13,7 @@
 #include <optional>
 #include <variant>
 
+#include <seastar/core/abort_source.hh>
 #include <seastar/core/sstring.hh>
 #include <seastar/core/timed_out_error.hh>
 
@@ -66,11 +67,14 @@ public:
     virtual const char* what() const noexcept override { return _message.c_str(); }
 };
 
+using abort_requested_exception = seastar::abort_requested_exception;
+
 struct exception_variant {
     std::variant<unknown_exception,
             no_exception,
             rate_limit_exception,
-            stale_topology_exception
+            stale_topology_exception,
+            abort_requested_exception
     > reason;
 
     exception_variant()

service/storage_proxy.cc

@@ -596,6 +596,9 @@ private:
                     } else if constexpr(std::is_same_v<Ex, replica::stale_topology_exception>) {
                         msg = e.what();
                         return error::FAILURE;
+                    } else if constexpr (std::is_same_v<Ex, replica::abort_requested_exception>) {
+                        msg = e.what();
+                        return error::FAILURE;
                     }
                 }, exception->reason);
             }