From 99379758291a15ee46f6df572ef0b0cfb1a6d9aa Mon Sep 17 00:00:00 2001
From: Nicolae Vartolomei <nv@redpanda.com>
Date: Fri, 27 Sep 2024 17:04:03 +0100
Subject: [PATCH] ssx: exit early from sleep_abortable if already aborted

Caught a few shutdown hangs in unit tests which I tracked down to
sleep_abortable being invoked with high durations after abort sources
where already aborted.

Do not try to sleep if abort source is already set. ss::sleep_abortable
behaves similarly from what I can read.

```
sleeper(typename Clock::duration dur, abort_source& as)
...
auto sc_opt = as.subscribe // returns empty optional if abort source is
already set
...
if (sc_opt) {...} else {
    done.set_exception(sleep_aborted());
}
```
---
 src/v/ssx/sleep_abortable.h             |  5 ++++
 src/v/ssx/tests/sleep_abortable_test.cc | 32 +++++++++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/src/v/ssx/sleep_abortable.h b/src/v/ssx/sleep_abortable.h
index af9b8f35d8207..7d76e64124c14 100644
--- a/src/v/ssx/sleep_abortable.h
+++ b/src/v/ssx/sleep_abortable.h
@@ -44,6 +44,11 @@ sleep_abortable(typename Clock::duration dur, AbortSource&... src) {
         }
         seastar::weak_ptr<as_state> state;
     };
+
+    if ((src.abort_requested() || ...)) {
+        return seastar::make_exception_future<>(seastar::sleep_aborted());
+    }
+
     auto state = seastar::make_lw_shared<as_state>();
     state->subscriptions.reserve(sizeof...(AbortSource));
     as_callback cb(*state);
diff --git a/src/v/ssx/tests/sleep_abortable_test.cc b/src/v/ssx/tests/sleep_abortable_test.cc
index 9611df6523913..67a2a037feb4a 100644
--- a/src/v/ssx/tests/sleep_abortable_test.cc
+++ b/src/v/ssx/tests/sleep_abortable_test.cc
@@ -31,6 +31,14 @@ SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort1) {
     BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
 }
 
+SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort1_pre) {
+    ss::abort_source as;
+
+    as.request_abort();
+    auto fut = ssx::sleep_abortable(1000s, as);
+    BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
+}
+
 SEASTAR_THREAD_TEST_CASE(sleep_abortable_normal1) {
     ss::abort_source as;
     ssx::sleep_abortable(10ms, as).get();
@@ -51,6 +59,15 @@ SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort2) {
     BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
 }
 
+SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort2_pre) {
+    ss::abort_source as1;
+    ss::abort_source as2;
+
+    as1.request_abort();
+    auto fut = ssx::sleep_abortable(1000s, as1, as2);
+    BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
+}
+
 SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort3) {
     ss::abort_source as1, as2, as3;
 
@@ -60,6 +77,14 @@ SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort3) {
     BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
 }
 
+SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort3_pre) {
+    ss::abort_source as1, as2, as3;
+
+    as3.request_abort();
+    auto fut = ssx::sleep_abortable(1000s, as1, as2, as3);
+    BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
+}
+
 SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort4) {
     ss::abort_source as1, as2, as3, as4;
     auto fut = ssx::sleep_abortable(1000s, as1, as2, as3, as4);
@@ -68,6 +93,13 @@ SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort4) {
     BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
 }
 
+SEASTAR_THREAD_TEST_CASE(sleep_abortable_abort4_pre) {
+    ss::abort_source as1, as2, as3, as4;
+    as2.request_abort();
+    auto fut = ssx::sleep_abortable(1000s, as1, as2, as3, as4);
+    BOOST_REQUIRE_THROW(fut.get(), ss::sleep_aborted);
+}
+
 SEASTAR_THREAD_TEST_CASE(sleep_abortable_aborted_twice) {
     ss::abort_source as1, as2;
     auto fut = ssx::sleep_abortable(1000s, as1, as2);