diff --git a/configure.py b/configure.py
index da90966366..901e58c92e 100755
--- a/configure.py
+++ b/configure.py
@@ -508,6 +508,8 @@ scylla_tests = set([
     'test/boost/rate_limiter_test',
     'test/boost/per_partition_rate_limit_test',
     'test/boost/expr_test',
+    'test/boost/exceptions_optimized_test',
+    'test/boost/exceptions_fallback_test',
     'test/manual/ec2_snitch_test',
     'test/manual/enormous_table_scan_test',
     'test/manual/gce_snitch_test',
@@ -1290,6 +1292,8 @@ deps['test/boost/linearizing_input_stream_test'] = [
 ]
 deps['test/boost/expr_test'] = ['test/boost/expr_test.cc'] + scylla_core
 deps['test/boost/rate_limiter_test'] = ['test/boost/rate_limiter_test.cc', 'db/rate_limiter.cc']
+deps['test/boost/exceptions_optimized_test'] = ['test/boost/exceptions_optimized_test.cc', 'utils/exceptions.cc']
+deps['test/boost/exceptions_fallback_test'] = ['test/boost/exceptions_fallback_test.cc', 'utils/exceptions.cc']
 
 deps['test/boost/duration_test'] += ['test/lib/exception_utils.cc']
 deps['test/boost/schema_loader_test'] += ['tools/schema_loader.cc']
diff --git a/message/rpc_protocol_impl.hh b/message/rpc_protocol_impl.hh
index 940e7d186a..653009e0ec 100644
--- a/message/rpc_protocol_impl.hh
+++ b/message/rpc_protocol_impl.hh
@@ -7,6 +7,7 @@
 #include "messaging_service.hh"
 #include "serializer.hh"
 #include "seastarx.hh"
+#include "utils/exceptions.hh"
 
 namespace netw {
 
@@ -123,27 +124,21 @@ void register_handler(messaging_service *ms, messaging_verb verb, Func &&func) {
 template <typename MsgIn, typename... MsgOut>
 auto send_message(messaging_service* ms, messaging_verb verb, msg_addr id, MsgOut&&... msg) {
     auto rpc_handler = ms->rpc()->make_client<MsgIn(MsgOut...)>(verb);
+    using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
     if (ms->is_shutting_down()) {
-        using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
         return futurator::make_exception_future(rpc::closed_error());
     }
     auto rpc_client_ptr = ms->get_rpc_client(verb, id);
     auto& rpc_client = *rpc_client_ptr;
-    return rpc_handler(rpc_client, std::forward<MsgOut>(msg)...).then_wrapped([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr)] (auto&& f) {
-        try {
-            if (f.failed()) {
-                ms->increment_dropped_messages(verb);
-                f.get();
-                assert(false); // never reached
-            }
-            return std::move(f);
-        } catch (rpc::closed_error&) {
+    return rpc_handler(rpc_client, std::forward<MsgOut>(msg)...).handle_exception([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr)] (std::exception_ptr&& eptr) {
+        ms->increment_dropped_messages(verb);
+        if (try_catch<rpc::closed_error>(eptr)) {
             // This is a transport error
             ms->remove_error_rpc_client(verb, id);
-            throw;
-        } catch (...) {
+            return futurator::make_exception_future(std::move(eptr));
+        } else {
             // This is expected to be a rpc server error, e.g., the rpc handler throws a std::runtime_error.
-            throw;
+            return futurator::make_exception_future(std::move(eptr));
         }
     });
 }
@@ -152,27 +147,21 @@ auto send_message(messaging_service* ms, messaging_verb verb, msg_addr id, MsgOu
 template <typename MsgIn, typename Timeout, typename... MsgOut>
 auto send_message_timeout(messaging_service* ms, messaging_verb verb, msg_addr id, Timeout timeout, MsgOut&&... msg) {
     auto rpc_handler = ms->rpc()->make_client<MsgIn(MsgOut...)>(verb);
+    using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
     if (ms->is_shutting_down()) {
-        using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
         return futurator::make_exception_future(rpc::closed_error());
     }
     auto rpc_client_ptr = ms->get_rpc_client(verb, id);
     auto& rpc_client = *rpc_client_ptr;
-    return rpc_handler(rpc_client, timeout, std::forward<MsgOut>(msg)...).then_wrapped([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr)] (auto&& f) {
-        try {
-            if (f.failed()) {
-                ms->increment_dropped_messages(verb);
-                f.get();
-                assert(false); // never reached
-            }
-            return std::move(f);
-        } catch (rpc::closed_error&) {
+    return rpc_handler(rpc_client, timeout, std::forward<MsgOut>(msg)...).handle_exception([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr)] (std::exception_ptr&& eptr) {
+        ms->increment_dropped_messages(verb);
+        if (try_catch<rpc::closed_error>(eptr)) {
             // This is a transport error
             ms->remove_error_rpc_client(verb, id);
-            throw;
-        } catch (...) {
+            return futurator::make_exception_future(std::move(eptr));
+        } else {
             // This is expected to be a rpc server error, e.g., the rpc handler throws a std::runtime_error.
-            throw;
+            return futurator::make_exception_future(std::move(eptr));
         }
     });
 }
@@ -183,8 +172,8 @@ auto send_message_timeout(messaging_service* ms, messaging_verb verb, msg_addr i
 template <typename MsgIn, typename... MsgOut>
 auto send_message_cancellable(messaging_service* ms, messaging_verb verb, msg_addr id, abort_source& as, MsgOut&&... msg) {
     auto rpc_handler = ms->rpc()->make_client<MsgIn(MsgOut...)>(verb);
+    using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
     if (ms->is_shutting_down()) {
-        using futurator = futurize<std::result_of_t<decltype(rpc_handler)(rpc_protocol::client&, MsgOut...)>>;
         return futurator::make_exception_future(rpc::closed_error());
     }
     auto rpc_client_ptr = ms->get_rpc_client(verb, id);
@@ -196,27 +185,21 @@ auto send_message_cancellable(messaging_service* ms, messaging_verb verb, msg_ad
         c->cancel();
     });
     if (!sub) {
-        throw abort_requested_exception{};
+        return futurator::make_exception_future(abort_requested_exception{});
     }
 
-    return rpc_handler(rpc_client, c_ref, std::forward<MsgOut>(msg)...).then_wrapped([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr), sub = std::move(sub)] (auto&& f) {
-        try {
-            if (f.failed()) {
-                ms->increment_dropped_messages(verb);
-                f.get();
-                assert(false); // never reached
-            }
-            return std::move(f);
-        } catch (rpc::closed_error&) {
+    return rpc_handler(rpc_client, c_ref, std::forward<MsgOut>(msg)...).handle_exception([ms = ms->shared_from_this(), id, verb, rpc_client_ptr = std::move(rpc_client_ptr), sub = std::move(sub)] (std::exception_ptr&& eptr) {
+        ms->increment_dropped_messages(verb);
+        if (try_catch<rpc::closed_error>(eptr)) {
             // This is a transport error
             ms->remove_error_rpc_client(verb, id);
-            throw;
-        } catch (rpc::canceled_error&) {
+            return futurator::make_exception_future(std::move(eptr));
+        } else if (try_catch<rpc::canceled_error>(eptr)) {
             // Translate low-level canceled_error into high-level abort_requested_exception.
-            throw abort_requested_exception{};
-        } catch (...) {
+            return futurator::make_exception_future(abort_requested_exception{});
+        } else {
             // This is expected to be a rpc server error, e.g., the rpc handler throws a std::runtime_error.
-            throw;
+            return futurator::make_exception_future(std::move(eptr));
         }
     });
 }
diff --git a/replica/database.cc b/replica/database.cc
index 7525243a0d..2de36c2fc5 100644
--- a/replica/database.cc
+++ b/replica/database.cc
@@ -21,6 +21,7 @@
 #include <seastar/core/seastar.hh>
 #include <seastar/core/coroutine.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
+#include <seastar/coroutine/as_future.hh>
 #include <seastar/core/reactor.hh>
 #include <seastar/core/metrics.hh>
 #include <boost/algorithm/string/erase.hpp>
@@ -1465,17 +1466,21 @@ database::query(schema_ptr s, const query::read_command& cmd, query::result_opti
     try {
         auto op = cf.read_in_progress();
 
+        future<> f = make_ready_future<>();
         if (querier_opt) {
-            co_await semaphore.with_ready_permit(querier_opt->permit(), read_func);
+            f = co_await coroutine::as_future(semaphore.with_ready_permit(querier_opt->permit(), read_func));
         } else {
-            co_await semaphore.with_permit(s.get(), "data-query", cf.estimate_read_memory_cost(), timeout, read_func);
+            f = co_await coroutine::as_future(semaphore.with_permit(s.get(), "data-query", cf.estimate_read_memory_cost(), timeout, read_func));
         }
 
-        if (cmd.query_uuid != utils::UUID{} && querier_opt) {
-            _querier_cache.insert(cmd.query_uuid, std::move(*querier_opt), std::move(trace_state));
+        if (!f.failed()) {
+            if (cmd.query_uuid != utils::UUID{} && querier_opt) {
+                _querier_cache.insert(cmd.query_uuid, std::move(*querier_opt), std::move(trace_state));
+            }
+        } else {
+            ex = f.get_exception();
         }
     } catch (...) {
-        ++semaphore.get_stats().total_failed_reads;
         ex = std::current_exception();
     }
 
@@ -1483,6 +1488,7 @@ database::query(schema_ptr s, const query::read_command& cmd, query::result_opti
         co_await querier_opt->close();
     }
     if (ex) {
+        ++semaphore.get_stats().total_failed_reads;
         co_return coroutine::exception(std::move(ex));
     }
 
@@ -1526,18 +1532,22 @@ database::query_mutations(schema_ptr s, const query::read_command& cmd, const dh
     try {
         auto op = cf.read_in_progress();
 
+        future<> f = make_ready_future<>();
         if (querier_opt) {
-            co_await semaphore.with_ready_permit(querier_opt->permit(), read_func);
+            f = co_await coroutine::as_future(semaphore.with_ready_permit(querier_opt->permit(), read_func));
         } else {
-            co_await semaphore.with_permit(s.get(), "mutation-query", cf.estimate_read_memory_cost(), timeout, read_func);
+            f = co_await coroutine::as_future(semaphore.with_permit(s.get(), "mutation-query", cf.estimate_read_memory_cost(), timeout, read_func));
         }
 
-        if (cmd.query_uuid != utils::UUID{} && querier_opt) {
-            _querier_cache.insert(cmd.query_uuid, std::move(*querier_opt), std::move(trace_state));
+        if (!f.failed()) {
+            if (cmd.query_uuid != utils::UUID{} && querier_opt) {
+                _querier_cache.insert(cmd.query_uuid, std::move(*querier_opt), std::move(trace_state));
+            }
+        } else {
+            ex = f.get_exception();
         }
 
     } catch (...) {
-        ++semaphore.get_stats().total_failed_reads;
         ex = std::current_exception();
     }
 
@@ -1545,6 +1555,7 @@ database::query_mutations(schema_ptr s, const query::read_command& cmd, const dh
         co_await querier_opt->close();
     }
     if (ex) {
+        ++semaphore.get_stats().total_failed_reads;
         co_return coroutine::exception(std::move(ex));
     }
 
@@ -1854,27 +1865,38 @@ public:
 
 // see above (#9919)
 template<typename T = std::runtime_error>
-static void throw_commitlog_add_error(schema_ptr s, const frozen_mutation& m) {
+static std::exception_ptr wrap_commitlog_add_error(schema_ptr s, const frozen_mutation& m, std::exception_ptr eptr) {
     // it is tempting to do a full pretty print here, but the mutation is likely
     // humungous if we got an error, so just tell us where and pk...
-    std::throw_with_nested(T(format("Could not write mutation {}:{} ({}) to commitlog"
+    return make_nested_exception_ptr(T(format("Could not write mutation {}:{} ({}) to commitlog"
         , s->ks_name(), s->cf_name()
         , m.key()
-    )));
+    )), std::move(eptr));
 }
 
 future<> database::apply_with_commitlog(column_family& cf, const mutation& m, db::timeout_clock::time_point timeout) {
     db::rp_handle h;
     if (cf.commitlog() != nullptr && cf.durable_writes()) {
         auto fm = freeze(m);
+        std::exception_ptr ex;
         try {
             commitlog_entry_writer cew(m.schema(), fm, db::commitlog::force_sync::no);
-            h = co_await cf.commitlog()->add_entry(m.schema()->id(), cew, timeout);
-        } catch (timed_out_error&) {
-            // see above (#9919)
-            throw_commitlog_add_error<wrapped_timed_out_error>(cf.schema(), fm);
+            auto f_h = co_await coroutine::as_future(cf.commitlog()->add_entry(m.schema()->id(), cew, timeout));
+            if (!f_h.failed()) {
+                h = f_h.get();
+            } else {
+                ex = f_h.get_exception();
+            }
         } catch (...) {
-            throw_commitlog_add_error<>(cf.schema(), fm);
+            ex = std::current_exception();
+        }
+        if (ex) {
+            if (try_catch<timed_out_error>(ex)) {
+                ex = wrap_commitlog_add_error<wrapped_timed_out_error>(cf.schema(), fm, std::move(ex));
+            } else {
+                ex = wrap_commitlog_add_error<>(cf.schema(), fm, std::move(ex));
+            }
+            co_await coroutine::exception(std::move(ex));
         }
     }
     try {
@@ -1973,14 +1995,25 @@ future<> database::do_apply(schema_ptr s, const frozen_mutation& m, tracing::tra
     db::rp_handle h;
     auto cl = cf.commitlog();
     if (cl != nullptr && cf.durable_writes()) {
+        std::exception_ptr ex;
         try {
             commitlog_entry_writer cew(s, m, sync);
-            h = co_await cf.commitlog()->add_entry(uuid, cew, timeout);
-        } catch (timed_out_error&) {
-            // see above (#9919)
-            throw_commitlog_add_error<wrapped_timed_out_error>(cf.schema(), m);
+            auto f_h = co_await coroutine::as_future(cf.commitlog()->add_entry(uuid, cew, timeout));
+            if (!f_h.failed()) {
+                h = f_h.get();
+            } else {
+                ex = f_h.get_exception();
+            }
         } catch (...) {
-            throw_commitlog_add_error<>(s, m);
+            ex = std::current_exception();
+        }
+        if (ex) {
+            if (try_catch<timed_out_error>(ex)) {
+                ex = wrap_commitlog_add_error<wrapped_timed_out_error>(cf.schema(), m, std::move(ex));
+            } else {
+                ex = wrap_commitlog_add_error<>(s, m, std::move(ex));
+            }
+            co_await coroutine::exception(std::move(ex));
         }
     }
     try {
@@ -1999,12 +2032,8 @@ Future database::update_write_metrics(Future&& f) {
             auto ep = f.get_exception();
             if (is_timeout_exception(ep)) {
                 ++s->total_writes_timedout;
-            }
-            try {
-                std::rethrow_exception(ep);
-            } catch (replica::rate_limit_exception&) {
+            } else if (try_catch<replica::rate_limit_exception>(ep)) {
                 ++s->total_writes_rate_limited;
-            } catch (...) {
             }
             return futurize<Future>::make_exception_future(std::move(ep));
         }
diff --git a/service/storage_proxy.cc b/service/storage_proxy.cc
index 8f3683caa5..d4a8272dc7 100644
--- a/service/storage_proxy.cc
+++ b/service/storage_proxy.cc
@@ -94,6 +94,7 @@
 #include "utils/overloaded_functor.hh"
 #include "utils/result_try.hh"
 #include "utils/error_injection.hh"
+#include "utils/exceptions.hh"
 #include "replica/exceptions.hh"
 #include "db/operation_type.hh"
 
@@ -2896,24 +2897,22 @@ void storage_proxy::send_to_live_endpoints(storage_proxy::response_id_type respo
             ++stats.writes_errors.get_ep_stat(p->get_token_metadata_ptr()->get_topology(), coordinator);
             error err = error::FAILURE;
             std::optional<sstring> msg;
-            try {
-                std::rethrow_exception(eptr);
-            } catch (replica::rate_limit_exception&) {
+            if (try_catch<replica::rate_limit_exception>(eptr)) {
                 // There might be a lot of those, so ignore
                 err = error::RATE_LIMIT;
-            } catch(rpc::closed_error&) {
+            } else if (try_catch<rpc::closed_error>(eptr)) {
                 // ignore, disconnect will be logged by gossiper
-            } catch(seastar::gate_closed_exception&) {
+            } else if (try_catch<seastar::gate_closed_exception>(eptr)) {
                 // may happen during shutdown, ignore it
-            } catch(timed_out_error&) {
+            } else if (try_catch<timed_out_error>(eptr)) {
                 // from lmutate(). Ignore so that logs are not flooded
                 // database total_writes_timedout counter was incremented.
                 // It needs to be recorded that the timeout occurred locally though.
                 err = error::TIMEOUT;
-            } catch(db::virtual_table_update_exception& e) {
-                msg = e.grab_cause();
-            } catch(...) {
-                slogger.error("exception during mutation write to {}: {}", coordinator, std::current_exception());
+            } else if (auto* e = try_catch<db::virtual_table_update_exception>(eptr)) {
+                msg = e->grab_cause();
+            } else {
+                slogger.error("exception during mutation write to {}: {}", coordinator, eptr);
             }
             p->got_failure_response(response_id, coordinator, forward_size + 1, std::nullopt, err, std::move(msg));
         });
@@ -2987,31 +2986,29 @@ public:
     void error(gms::inet_address ep, std::exception_ptr eptr) {
         sstring why;
         error_kind kind = error_kind::FAILURE;
-        try {
-            std::rethrow_exception(eptr);
-        } catch (replica::rate_limit_exception&) {
+        if (auto ex = try_catch<replica::rate_limit_exception>(eptr)) {
             // There might be a lot of those, so ignore
             kind = error_kind::RATE_LIMIT;
-        } catch (rpc::closed_error&) {
+        } else if (auto ex = try_catch<rpc::closed_error>(eptr)) {
             // do not report connection closed exception, gossiper does that
             kind = error_kind::DISCONNECT;
-        } catch (rpc::timeout_error&) {
+        } else if (try_catch<rpc::timeout_error>(eptr)) {
             // do not report timeouts, the whole operation will timeout and be reported
             return; // also do not report timeout as replica failure for the same reason
-        } catch (semaphore_timed_out&) {
+        } else if (try_catch<semaphore_timed_out>(eptr)) {
             // do not report timeouts, the whole operation will timeout and be reported
             return; // also do not report timeout as replica failure for the same reason
-        } catch (timed_out_error&) {
+        } else if (try_catch<timed_out_error>(eptr)) {
             // do not report timeouts, the whole operation will timeout and be reported
             return; // also do not report timeout as replica failure for the same reason
-        } catch (abort_requested_exception& e) {
+        } else if (try_catch<abort_requested_exception>(eptr)) {
             // do not report aborts, they are trigerred by shutdown or timeouts
-        } catch (rpc::remote_verb_error& e) {
+        } else if (auto ex = try_catch<rpc::remote_verb_error>(eptr)) {
             // Log remote read error with lower severity.
             // If it is really severe it we be handled on the host that sent
             // it.
-            slogger.warn("Exception when communicating with {}, to read from {}.{}: {}", ep, _schema->ks_name(), _schema->cf_name(), e.what());
-        } catch(...) {
+            slogger.warn("Exception when communicating with {}, to read from {}.{}: {}", ep, _schema->ks_name(), _schema->cf_name(), ex->what());
+        } else {
             slogger.error("Exception when communicating with {}, to read from {}.{}: {}", ep, _schema->ks_name(), _schema->cf_name(), eptr);
         }
 
@@ -3749,16 +3746,24 @@ protected:
         for (const gms::inet_address& ep : boost::make_iterator_range(begin, end)) {
             // Waited on indirectly, shared_from_this keeps `this` alive
             (void)make_mutation_data_request(cmd, ep, timeout).then_wrapped([this, resolver, ep, start, exec = shared_from_this()] (future<rpc::tuple<foreign_ptr<lw_shared_ptr<reconcilable_result>>, cache_temperature>> f) {
+                std::exception_ptr ex;
                 try {
+                  if (!f.failed()) {
                     auto v = f.get0();
                     _cf->set_hit_rate(ep, std::get<1>(v));
                     resolver->add_mutate_data(ep, std::get<0>(std::move(v)));
                     ++_proxy->get_stats().mutation_data_read_completed.get_ep_stat(get_topology(), ep);
                     register_request_latency(latency_clock::now() - start);
-                } catch(...) {
-                    ++_proxy->get_stats().mutation_data_read_errors.get_ep_stat(get_topology(), ep);
-                    resolver->error(ep, std::current_exception());
+                    return;
+                  } else {
+                    ex = f.get_exception();
+                  }
+                } catch (...) {
+                  ex = std::current_exception();
                 }
+
+                ++_proxy->get_stats().mutation_data_read_errors.get_ep_stat(get_topology(), ep);
+                resolver->error(ep, std::move(ex));
             });
         }
     }
@@ -3767,17 +3772,25 @@ protected:
         for (const gms::inet_address& ep : boost::make_iterator_range(begin, end)) {
             // Waited on indirectly, shared_from_this keeps `this` alive
             (void)make_data_request(ep, timeout, want_digest).then_wrapped([this, resolver, ep, start, exec = shared_from_this()] (future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature>> f) {
+                std::exception_ptr ex;
                 try {
+                  if (!f.failed()) {
                     auto v = f.get0();
                     _cf->set_hit_rate(ep, std::get<1>(v));
                     resolver->add_data(ep, std::get<0>(std::move(v)));
                     ++_proxy->get_stats().data_read_completed.get_ep_stat(get_topology(), ep);
                     _used_targets.push_back(ep);
                     register_request_latency(latency_clock::now() - start);
-                } catch(...) {
-                    ++_proxy->get_stats().data_read_errors.get_ep_stat(get_topology(), ep);
-                    resolver->error(ep, std::current_exception());
+                    return;
+                  } else {
+                    ex = f.get_exception();
+                  }
+                } catch (...) {
+                  ex = std::current_exception();
                 }
+
+                ++_proxy->get_stats().data_read_errors.get_ep_stat(get_topology(), ep);
+                resolver->error(ep, std::move(ex));
             });
         }
     }
@@ -3786,17 +3799,25 @@ protected:
         for (const gms::inet_address& ep : boost::make_iterator_range(begin, end)) {
             // Waited on indirectly, shared_from_this keeps `this` alive
             (void)make_digest_request(ep, timeout).then_wrapped([this, resolver, ep, start, exec = shared_from_this()] (future<rpc::tuple<query::result_digest, api::timestamp_type, cache_temperature>> f) {
+                std::exception_ptr ex;
                 try {
+                  if (!f.failed()) {
                     auto v = f.get0();
                     _cf->set_hit_rate(ep, std::get<2>(v));
                     resolver->add_digest(ep, std::get<0>(v), std::get<1>(v));
                     ++_proxy->get_stats().digest_read_completed.get_ep_stat(get_topology(), ep);
                     _used_targets.push_back(ep);
                     register_request_latency(latency_clock::now() - start);
-                } catch(...) {
-                    ++_proxy->get_stats().digest_read_errors.get_ep_stat(get_topology(), ep);
-                    resolver->error(ep, std::current_exception());
+                    return;
+                  } else {
+                    ex = f.get_exception();
+                  }
+                } catch (...) {
+                  ex = std::current_exception();
                 }
+
+                ++_proxy->get_stats().digest_read_errors.get_ep_stat(get_topology(), ep);
+                resolver->error(ep, std::move(ex));
             });
         }
     }
diff --git a/test/boost/exceptions_fallback_test.cc b/test/boost/exceptions_fallback_test.cc
new file mode 100644
index 0000000000..9f4627a4a4
--- /dev/null
+++ b/test/boost/exceptions_fallback_test.cc
@@ -0,0 +1,11 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#define NO_OPTIMIZED_EXCEPTION_HANDLING
+
+#include "exceptions_test.inc.cc"
diff --git a/test/boost/exceptions_optimized_test.cc b/test/boost/exceptions_optimized_test.cc
new file mode 100644
index 0000000000..6db2b59148
--- /dev/null
+++ b/test/boost/exceptions_optimized_test.cc
@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#if defined(NO_OPTIMIZED_EXCEPTION_HANDLING)
+#undef NO_OPTIMIZED_EXCEPTION_HANDLING
+#endif
+
+#include "utils/exceptions.hh"
+
+#if defined(OPTIMIZED_EXCEPTION_HANDLING_AVAILABLE)
+
+#include "exceptions_test.inc.cc"
+
+#else
+
+#include <boost/test/unit_test_log.hpp>
+#include <seastar/testing/test_case.hh>
+
+SEASTAR_TEST_CASE(test_noop) {
+    BOOST_TEST_MESSAGE("Optimized implementation of handling exceptions "
+            "without throwing is not available. Skipping tests in this file.");
+    return seastar::make_ready_future<>();
+}
+
+#endif
diff --git a/test/boost/exceptions_test.inc.cc b/test/boost/exceptions_test.inc.cc
new file mode 100644
index 0000000000..de75ab7ac5
--- /dev/null
+++ b/test/boost/exceptions_test.inc.cc
@@ -0,0 +1,178 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+// Common definitions of test cases used in
+//   handle_exception_optimized_test.cc
+//   handle_exception_fallback_test.cc
+
+#include <exception>
+#include <stdexcept>
+#include <type_traits>
+#include <cxxabi.h>
+#include <boost/test/unit_test_log.hpp>
+#include <boost/test/tools/old/interface.hpp>
+#include <boost/test/unit_test.hpp>
+
+#include "seastarx.hh"
+#include <seastar/core/future.hh>
+#include <seastar/testing/test_case.hh>
+#include <seastar/util/log.hh>
+
+#include "utils/exceptions.hh"
+
+class base_exception : public std::exception {};
+class derived_exception : public base_exception {};
+
+static void dummy_fn(void*) {
+    //
+}
+
+template<typename T>
+static std::exception_ptr maybe_wrap_eptr(T&& t) {
+    if constexpr (std::is_same_v<T, std::exception_ptr>) {
+        return std::move(t);
+    } else {
+        return std::make_exception_ptr(std::move(t));
+    }
+}
+
+static const std::type_info& eptr_typeid(std::exception_ptr eptr) {
+    try {
+        std::rethrow_exception(eptr);
+    } catch (...) {
+        return *abi::__cxa_current_exception_type();
+    }
+}
+
+template<typename Capture, typename Throw>
+static void check_catch(Throw&& ex) {
+    auto eptr = maybe_wrap_eptr(std::move(ex));
+    BOOST_TEST_MESSAGE("Checking if " << seastar::pretty_type_name(eptr_typeid(eptr))
+            << " is caught as " << seastar::pretty_type_name(typeid(Capture)));
+
+    auto typed_eptr = try_catch<Capture>(eptr);
+    BOOST_CHECK_NE(typed_eptr, nullptr);
+
+    // Verify that it's the same as what the usual throwing gives
+    // TODO: Does this check make sense? Does the standard guarantee
+    // that this will give the same pointer?
+    try {
+        std::rethrow_exception(eptr);
+    } catch (Capture& t) {
+        BOOST_CHECK_EQUAL(typed_eptr, &t);
+    } catch (...) {
+        // Can happen if the first check fails, just skip
+        assert(typed_eptr == nullptr);
+    }
+}
+
+template<typename Capture, typename Throw>
+static void check_no_catch(Throw&& ex) {
+    auto eptr = maybe_wrap_eptr(std::move(ex));
+    BOOST_TEST_MESSAGE("Checking if " << seastar::pretty_type_name(eptr_typeid(eptr))
+            << " is NOT caught as " << seastar::pretty_type_name(typeid(Capture)));
+
+    auto typed_eptr = try_catch<Capture>(eptr);
+    BOOST_CHECK_EQUAL(typed_eptr, nullptr);
+}
+
+template<typename A, typename B>
+static std::exception_ptr make_nested_eptr(A&& a, B&& b) {
+    try {
+        throw std::move(b);
+    } catch (...) {
+        try {
+            std::throw_with_nested(std::move(a));
+        } catch (...) {
+            return std::current_exception();
+        }
+    }
+}
+
+SEASTAR_TEST_CASE(test_try_catch) {
+    // Some standard examples, throwing exceptions derived from std::exception
+    // and catching them through their base types
+
+    check_catch<derived_exception>(derived_exception());
+    check_catch<base_exception>(derived_exception());
+    check_catch<std::exception>(derived_exception());
+    check_no_catch<std::runtime_error>(derived_exception());
+
+    check_no_catch<derived_exception>(base_exception());
+    check_catch<base_exception>(base_exception());
+    check_catch<std::exception>(base_exception());
+    check_no_catch<std::runtime_error>(base_exception());
+
+    // Catching nested exceptions
+    check_catch<base_exception>(make_nested_eptr(base_exception(), derived_exception()));
+    check_catch<std::nested_exception>(make_nested_eptr(base_exception(), derived_exception()));
+    check_no_catch<derived_exception>(make_nested_eptr(base_exception(), derived_exception()));
+
+    // Check that everything works if we throw some crazy stuff
+    check_catch<int>(int(1));
+    check_no_catch<std::exception>(int(1));
+
+    check_no_catch<int>(nullptr);
+    check_no_catch<std::exception>(nullptr);
+
+    // Catching pointers is not supported, but nothing should break if they are
+    // being thrown
+    derived_exception exc;
+    check_no_catch<int>(&exc);
+    check_no_catch<std::exception>(&exc);
+
+    check_no_catch<int>(&dummy_fn);
+    check_no_catch<std::exception>(&dummy_fn);
+
+    check_no_catch<int>(&std::exception::what);
+    check_no_catch<std::exception>(&std::exception::what);
+
+    return make_ready_future<>();
+}
+
+SEASTAR_TEST_CASE(test_make_nested_exception_ptr) {
+    auto inner = std::make_exception_ptr(std::runtime_error("inner"));
+    auto outer = make_nested_exception_ptr(std::runtime_error("outer"), inner);
+
+    try {
+        std::rethrow_exception(outer);
+    } catch (const std::runtime_error& ex) {
+        BOOST_REQUIRE_EQUAL(std::string(ex.what()), "outer");
+        auto* nested = dynamic_cast<const std::nested_exception*>(&ex);
+        BOOST_REQUIRE_NE(nested, nullptr);
+        BOOST_REQUIRE_EQUAL(nested->nested_ptr(), inner);
+    }
+
+    try {
+        std::rethrow_exception(outer);
+    } catch (const std::nested_exception& ex) {
+        BOOST_REQUIRE_EQUAL(ex.nested_ptr(), inner);
+    }
+
+    // Not a class
+    BOOST_REQUIRE_THROW(std::rethrow_exception(make_nested_exception_ptr(int(123), inner)), int);
+
+    // Final, so cannot add the std::nested_exception mixin to it
+    struct ultimate_exception final : public std::exception {};
+    BOOST_REQUIRE_THROW(std::rethrow_exception(make_nested_exception_ptr(ultimate_exception(), inner)), ultimate_exception);
+
+    // Already derived from nested exception, so cannot put more to it
+    struct already_nested_exception : public std::exception, public std::nested_exception {};
+    auto inner2 = std::make_exception_ptr(std::runtime_error("inner2"));
+    try {
+        std::rethrow_exception(inner2);
+    } catch (const std::runtime_error&) {
+        try {
+            std::rethrow_exception(make_nested_exception_ptr(already_nested_exception(), inner));
+        } catch (const already_nested_exception& ex) {
+            BOOST_REQUIRE_EQUAL(ex.nested_ptr(), inner2);
+        }
+    }
+
+    return make_ready_future<>();
+}
diff --git a/utils/abi/eh_ia64.hh b/utils/abi/eh_ia64.hh
new file mode 100644
index 0000000000..2ecd8b4715
--- /dev/null
+++ b/utils/abi/eh_ia64.hh
@@ -0,0 +1,52 @@
+
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#pragma once
+
+#include <unwind.h>
+#include <typeinfo>
+#include <exception>
+
+// This file defines structures/functions derived from the Itanium C++ ABI.
+// Source: https://itanium-cxx-abi.github.io/cxx-abi/abi-eh.html
+
+namespace utils {
+namespace abi {
+
+// __cxa_exception (exception object header), as defined in section 2.2.1
+struct cxa_exception { 
+    std::type_info* exceptionType;
+    void (*exceptionDestructor)(void*); 
+    std::unexpected_handler unexpectedHandler;
+    std::terminate_handler terminateHandler;
+    cxa_exception* nextException;
+
+    int handlerCount;
+    int	handlerSwitchValue;
+    const char* actionRecord;
+    const char* languageSpecificData;
+    void* catchTemp;
+    void* adjustedPtr;
+
+    _Unwind_Exception unwindHeader;
+};
+
+// Given a pointer to the exception data, returns the pointer
+// to the __cxa_exception header.
+inline cxa_exception* get_cxa_exception(void* eptr) {
+    // From section 2.2.1:
+    // "By convention, a __cxa_exception pointer points at the C++ object
+    // representing the exception being thrown, immediately following
+    // the header. The header structure is accessed at a negative offset
+    // from the __cxa_exception pointer."
+    return reinterpret_cast<cxa_exception*>(eptr) - 1;
+}
+
+} // abi
+} // utils
diff --git a/utils/exceptions.cc b/utils/exceptions.cc
index 0e8006cb1b..f3d424cf20 100644
--- a/utils/exceptions.cc
+++ b/utils/exceptions.cc
@@ -15,6 +15,7 @@
 #include <system_error>
 #include <atomic>
 #include "exceptions.hh"
+#include "utils/abi/eh_ia64.hh"
 
 #include <iostream>
 
@@ -59,17 +60,36 @@ bool should_stop_on_system_error(const std::system_error& e) {
 }
 
 bool is_timeout_exception(std::exception_ptr e) {
-    try {
-        std::rethrow_exception(e);
-    } catch (seastar::rpc::timeout_error& unused) {
+    if (try_catch<seastar::rpc::timeout_error>(e)) {
         return true;
-    } catch (seastar::semaphore_timed_out& unused) {
+    } else if (try_catch<seastar::semaphore_timed_out>(e)) {
         return true;
-    } catch (seastar::timed_out_error& unused) {
+    } else if (try_catch<seastar::timed_out_error>(e)) {
         return true;
-    } catch (const std::nested_exception& e) {
-        return is_timeout_exception(e.nested_ptr());
-    } catch (...) {
+    } else if (const auto* ex = try_catch<const std::nested_exception>(e)) {
+        return is_timeout_exception(ex->nested_ptr());
     }
     return false;
 }
+
+#if defined(OPTIMIZED_EXCEPTION_HANDLING_AVAILABLE)
+
+#include <typeinfo>
+#include "utils/abi/eh_ia64.hh"
+
+void* utils::internal::try_catch_dynamic(std::exception_ptr& eptr, const std::type_info* catch_type) noexcept {
+    // In both libstdc++ and libc++, exception_ptr has just one field
+    // which is a pointer to the exception data
+    void* raw_ptr = reinterpret_cast<void*&>(eptr);
+    const std::type_info* ex_type = utils::abi::get_cxa_exception(raw_ptr)->exceptionType;
+
+    // __do_catch can return true and set raw_ptr to nullptr, but only in the case
+    // when catch_type is a pointer and a nullptr is thrown. try_catch_dynamic
+    // doesn't work with catching pointers.
+    if (catch_type->__do_catch(ex_type, &raw_ptr, 1)) {
+        return raw_ptr;
+    }
+    return nullptr;
+}
+
+#endif // __GLIBCXX__
diff --git a/utils/exceptions.hh b/utils/exceptions.hh
index 83714e5af6..b58b7b16d3 100644
--- a/utils/exceptions.hh
+++ b/utils/exceptions.hh
@@ -8,11 +8,27 @@
 
 #pragma once
 
+#include <cstddef>
+
+#if defined(__GLIBCXX__) && (defined(__x86_64__) || defined(__aarch64__))
+  #define OPTIMIZED_EXCEPTION_HANDLING_AVAILABLE
+#endif
+
+#if !defined(NO_OPTIMIZED_EXCEPTION_HANDLING)
+  #if defined(OPTIMIZED_EXCEPTION_HANDLING_AVAILABLE)
+    #define USE_OPTIMIZED_EXCEPTION_HANDLING
+  #else
+    #warn "Fast implementation of some of the exception handling routines is not available for this platform. Expect poor exception handling performance."
+  #endif
+#endif
+
 #include <seastar/core/sstring.hh>
 #include <seastar/core/on_internal_error.hh>
+#include <seastar/core/align.hh>
 
 #include <functional>
 #include <system_error>
+#include <type_traits>
 
 namespace seastar { class logger; }
 
@@ -60,3 +76,106 @@ inline void maybe_rethrow_exception(std::exception_ptr ex) {
         std::rethrow_exception(std::move(ex));
     }
 }
+
+namespace utils::internal {
+
+#if defined(OPTIMIZED_EXCEPTION_HANDLING_AVAILABLE)
+void* try_catch_dynamic(std::exception_ptr& eptr, const std::type_info* catch_type) noexcept;
+
+template<typename Ex>
+class nested_exception : public Ex, public std::nested_exception {
+private:
+    void set_nested_exception(std::exception_ptr nested_eptr) {
+        // Hack: libstdc++'s std::nested_exception has just one field
+        // which is a std::exception_ptr. It is initialized
+        // to std::current_exception on its construction, but we override
+        // it here.
+        auto* nex = dynamic_cast<std::nested_exception*>(this);
+
+        // std::nested_exception is virtual without any base classes,
+        // so according to the ABI we just need to skip the vtable pointer
+        // and align
+        auto* nptr = reinterpret_cast<std::exception_ptr*>(
+                seastar::align_up(
+                        reinterpret_cast<char*>(nex) + sizeof(void*),
+                        alignof(std::nested_exception)));
+        *nptr = std::move(nested_eptr);
+    }
+
+public:
+    explicit nested_exception(const Ex& ex, std::exception_ptr&& nested_eptr)
+            : Ex(ex) {
+        set_nested_exception(std::move(nested_eptr));
+    }
+
+    explicit nested_exception(Ex&& ex, std::exception&& nested_eptr)
+            : Ex(std::move(ex)) {
+        set_nested_exception(std::move(nested_eptr));
+    }
+};
+#endif
+
+} // utils::internal
+
+/// If the exception_ptr holds an exception which would match on a `catch (T&)`
+/// clause, returns a pointer to it. Otherwise, returns `nullptr`.
+///
+/// The exception behind the pointer is valid as long as the exception
+/// behind the exception_ptr is valid.
+template<typename T>
+inline T* try_catch(std::exception_ptr& eptr) noexcept {
+    static_assert(!std::is_pointer_v<T>, "catching pointers is not supported");
+    static_assert(!std::is_lvalue_reference_v<T> && !std::is_rvalue_reference_v<T>,
+            "T must not be a reference");
+
+#if defined(USE_OPTIMIZED_EXCEPTION_HANDLING)
+    void* opt_ptr = utils::internal::try_catch_dynamic(eptr, &typeid(std::remove_const_t<T>));
+    return reinterpret_cast<T*>(opt_ptr);
+#else
+    try {
+        std::rethrow_exception(eptr);
+    } catch (T& t) {
+        return &t;
+    } catch (...) {
+    }
+    return nullptr;
+#endif
+}
+
+/// Analogous to std::throw_with_nested, but instead of capturing the currently
+/// thrown exception, takes the exception to be nested inside as an argument,
+/// and does not throw the new exception but rather returns it.
+template<typename Ex>
+inline std::exception_ptr make_nested_exception_ptr(Ex&& ex, std::exception_ptr nested) {
+    using ExDecayed = std::decay_t<Ex>;
+
+    static_assert(std::is_copy_constructible_v<ExDecayed> && std::is_move_constructible_v<ExDecayed>,
+            "make_nested_exception_ptr argument must be CopyConstructible");
+
+#if defined(USE_OPTIMIZED_EXCEPTION_HANDLING)
+    // std::rethrow_with_nested wraps the exception type if and only if
+    // it is a non-final non-union class type
+    // and is neither std::nested_exception nor derived from it.
+    // Ref: https://en.cppreference.com/w/cpp/error/throw_with_nested
+    constexpr bool wrap = std::is_class_v<ExDecayed>
+            && !std::is_final_v<ExDecayed>
+            && !std::is_base_of_v<std::nested_exception, ExDecayed>;
+
+    if constexpr (wrap) {
+        return std::make_exception_ptr(utils::internal::nested_exception<ExDecayed>(
+                std::forward<Ex>(ex), std::move(nested)));
+    } else {
+        return std::make_exception_ptr<Ex>(std::forward<Ex>(ex));
+    }
+#else
+    try {
+        std::rethrow_exception(std::move(nested));
+    } catch (...) {
+        try {
+            std::throw_with_nested(std::forward<Ex>(ex));
+        } catch (...) {
+            return std::current_exception();
+        }
+    }
+#endif
+}