core: sleep_abortable propagates custom abort exception when already aborted

sleep_abortable(dur, abort_source&) had two abort paths that disagreed
on the failure exception. Aborting during the sleep propagated a custom
exception set via request_abort_ex, but if the source was already
aborted before sleep_abortable subscribed, subscribe() returned a
disengaged optional and the fast path always fabricated a fresh
sleep_aborted, discarding the custom exception. The resulting exception
depended on a race with the subscribe call.

Make the already-aborted fast path mirror the during-sleep path:
propagate the caller-supplied exception when one was given, otherwise
fall back to sleep_aborted. abort_source::get_default_exception() now
returns a stable shared exception_ptr, which a plain request_abort()
stores; sleep_abortable distinguishes a caller-supplied exception by
comparing the stored abort exception against it by identity. A plain
request_abort() still yields sleep_aborted as before.

This relies on comparing exception_ptr identity, so abort_source is
changed to return its default exception_ptr derived from a common
thread_local initial value.

Closes scylladb#3452

Author: travisdowns

include/seastar/core/abort_source.hh

@@ -213,8 +213,15 @@ public:
 
     /// Returns the default exception type (\ref abort_requested_exception) for this abort source.
     /// Overridable by derived classes.
+    ///
+    /// The same exception_ptr object is returned on every call, so callers can
+    /// compare an abort exception (\ref abort_requested_exception_ptr) against
+    /// it by identity to tell whether an abort used the default exception or a
+    /// caller-supplied one. Overrides should preserve this property by likewise
+    /// returning a stable object.
     virtual std::exception_ptr get_default_exception() const noexcept {
-        return make_exception_ptr(abort_requested_exception());
+        static const thread_local std::exception_ptr ex = make_exception_ptr(abort_requested_exception());
+        return ex;
     }
 };
 

include/seastar/core/sleep.hh

@@ -77,12 +77,16 @@ future<> sleep_abortable(typename Clock::duration dur);
 extern template future<> sleep_abortable<steady_clock_type>(typename steady_clock_type::duration);
 extern template future<> sleep_abortable<lowres_clock>(typename lowres_clock::duration);
 
-/// Returns a future which completes after a specified time has elapsed
-/// or throws \ref sleep_aborted exception if the sleep is aborted.
+/// Returns a future which completes after a specified time has elapsed,
+/// or returns a failed future immediately if the sleep is aborted via \p as.
+///
+/// If the abort carried a custom exception (via \ref abort_source::request_abort_ex),
+/// the returned future is failed with that exception; otherwise it is failed
+/// with \ref sleep_aborted.
 ///
 /// \param dur minimum amount of time before the returned future becomes
 ///            ready.
-/// \param as the \ref abort_source that eventually notifies that the sleep
+/// \param as the \ref abort_source that can be used to notify that the sleep
 ///            should be aborted.
 /// \return A \ref future which becomes ready when the sleep duration elapses.
 template <typename Clock = steady_clock_type>

src/core/future-util.cc

@@ -111,7 +111,16 @@ future<> sleep_abortable(typename Clock::duration dur, abort_source& as) {
                 sc = std::move(*sc_opt);
                 tmr.arm(dur);
             } else {
-                done.set_exception(sleep_aborted());
+                // Try to preserve a custom exception if one was provided, relying on
+                // comparing the exception pointers. This relies on get_default_exception
+                // always providing a pointer to the same exception object.
+                // See scylladb/seastar#3452.
+                const auto& ex = as.abort_requested_exception_ptr();
+                if (ex != as.get_default_exception()) {
+                    done.set_exception(ex);
+                } else {
+                    done.set_exception(sleep_aborted());
+                }
             }
         }
     };

tests/unit/abort_source_test.cc

@@ -201,6 +201,23 @@ SEASTAR_THREAD_TEST_CASE(test_sleep_abortable_already_aborted_with_exception) {
     BOOST_REQUIRE(caught_exception);
 }
 
+// Companion to the above: when the already-aborted source carries no custom
+// exception (plain request_abort), sleep_abortable still fails with
+// sleep_aborted, matching the during-sleep path. See scylladb/seastar#3452.
+SEASTAR_THREAD_TEST_CASE(test_sleep_abortable_already_aborted_no_exception) {
+    abort_source as;
+    as.request_abort();
+    auto f = sleep_abortable(10s, as);
+
+    bool caught_exception = false;
+    try {
+        f.get();
+    } catch (const sleep_aborted&) {
+        caught_exception = true;
+    }
+    BOOST_REQUIRE(caught_exception);
+}
+
 SEASTAR_THREAD_TEST_CASE(test_destroy_with_moved_subscriptions) {
     auto as = std::make_unique<abort_source>();
     int aborted = 0;