fix: timeout callbacks hold their session, not this

include/anyhttp/asio/runtime.hpp

@@ -521,8 +521,13 @@ class Event
 /**
  * A timer that calls back when it expires, which is what an engine with timeouts of its own
  * (ngtcp2) needs: it tells when, and wants to be called then. Arming the timer again replaces
- * both the expiry and the callback, and a callback that has been replaced or cancelled is never
- * called. Callbacks run on the timer's executor.
+ * both the expiry and the callback. Callbacks run on the timer's executor.
+ *
+ * Arming it again, cancel() and the destructor stop a wait that is still pending, but no
+ * asynchronous operation can be taken back once it has completed: when the delay has expired
+ * already, the callback is on its way, and it still runs, once, soon after. So a callback holds
+ * what it uses rather than refer to it -- a `weak_ptr` to its session, not `this` -- and a call
+ * that is late does no harm.
  */
 class Timer
 {

include/anyhttp/h2/session_details.hpp

@@ -235,7 +235,8 @@ Task<void> NGHttp2SessionImpl<Stream>::end_connection()
    //
    if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec)
    {
-      if (auto drained = co_await io::drain(stream_, this->get_executor(), std::chrono::seconds(2)))
+      if (auto drained = co_await io::drain(stream_, this->weak_from_this(), this->get_executor(),
+                                            std::chrono::seconds(2)))
          mlogd("dropped {} bytes the peer sent after the session ended", drained);
    }
    else if (ec != errc::not_connected) // the peer may be gone already
@@ -456,7 +457,11 @@ IoTask<> ClientSession<Stream>::shutdown()
    this->start_write();
 
    Timer timer(this->get_executor());
-   timer.arm(std::chrono::seconds(2), [this] { io::cancel(stream_); });
+   timer.arm(std::chrono::seconds(2), [weak = this->weak_from_this()] {
+      // may come late, see Timer
+      if (auto self = std::static_pointer_cast<ClientSession>(weak.lock()))
+         io::cancel(self->stream_);
+   });
    co_await this->session_done_.wait();
    timer.cancel();
 

include/anyhttp/h2v2/session_details.hpp

@@ -234,7 +234,8 @@ Task<void> NGHttp2SessionImpl<Stream>::end_connection()
    //
    if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec)
    {
-      if (auto drained = co_await io::drain(stream_, this->get_executor(), std::chrono::seconds(2)))
+      if (auto drained = co_await io::drain(stream_, this->weak_from_this(), this->get_executor(),
+                                            std::chrono::seconds(2)))
          mlogd("dropped {} bytes the peer sent after the session ended", drained);
    }
    else if (ec != errc::not_connected) // the peer may be gone already
@@ -388,7 +389,11 @@ IoTask<> ClientSession<Stream>::shutdown()
    this->start_write();
 
    Timer timer(this->get_executor());
-   timer.arm(std::chrono::seconds(2), [this] { io::cancel(stream_); });
+   timer.arm(std::chrono::seconds(2), [weak = this->weak_from_this()] {
+      // may come late, see Timer
+      if (auto self = std::static_pointer_cast<ClientSession>(weak.lock()))
+         io::cancel(self->stream_);
+   });
    co_await this->session_done_.wait();
    timer.cancel();
 

include/anyhttp/net.hpp

@@ -43,6 +43,7 @@
 #include <array>
 #include <cerrno>
 #include <chrono>
+#include <memory>
 #include <optional>
 #include <string_view>
 
@@ -70,13 +71,20 @@ namespace anyhttp::inline asio::io
  *
  * The data is of no interest, not even decrypted: this reads the socket itself, by hand, as that
  * is what every stream type has in common.
+ *
+ * \p owner is what keeps \p stream alive: a timeout that expires as the peer ends its side may
+ * still cancel the stream after this has returned (see Timer), and does nothing once \p owner is
+ * gone.
  */
 template <SocketStream Stream>
-Task<size_t> drain(Stream& stream, const Executor& executor,
+Task<size_t> drain(Stream& stream, std::weak_ptr<const void> owner, const Executor& executor,
                    std::chrono::steady_clock::duration timeout)
 {
    Timer timer(executor);
-   timer.arm(timeout, [&stream] { io::cancel(stream); });
+   timer.arm(timeout, [&stream, weak = std::move(owner)] {
+      if (auto self = weak.lock())
+         io::cancel(stream);
+   });
 
    auto& socket = get_socket(stream);
    std::array<char, 16 * 1024> buffer;

src/h1/session.cpp

@@ -866,7 +866,7 @@ Task<void> BeastSession<Stream>::end_connection(bool notify)
    //
    if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec)
    {
-      if (auto drained = co_await io::drain(stream_, get_executor(), 2s))
+      if (auto drained = co_await io::drain(stream_, this->weak_from_this(), get_executor(), 2s))
          mlogd("dropped {} bytes the peer sent after the last message", drained);
    }
    else if (ec != errc::not_connected) // the peer may be gone already

test/test_corosio.cpp

@@ -422,21 +422,6 @@ TEST(CorosioRuntime, WHEN_event_is_set_before_and_while_waiting_THEN_waits_end)
    EXPECT_EQ(woken, 2);
 }
 
-TEST(CorosioRuntime, WHEN_timer_is_rearmed_or_cancelled_THEN_only_the_last_callback_runs)
-{
-   boost::corosio::io_context context;
-   Timer timer(context.get_executor());
-   std::vector<int> fired;
-
-   timer.arm(std::chrono::milliseconds(1), [&] { fired.push_back(1); });
-   timer.arm(std::chrono::milliseconds(2), [&] { fired.push_back(2); });
-   Timer other(context.get_executor());
-   other.arm(std::chrono::milliseconds(1), [&] { fired.push_back(3); });
-   other.cancel();
-   context.run();
-   EXPECT_THAT(fired, ElementsAre(2));
-}
-
 TEST(CorosioRuntime, WHEN_both_tasks_run_THEN_when_both_waits_for_both)
 {
    boost::corosio::io_context context;

test/test_runtime.cpp

@@ -22,6 +22,7 @@
 #include <chrono>
 #include <exception>
 #include <optional>
+#include <vector>
 
 using namespace anyhttp;
 using namespace std::chrono_literals;
@@ -165,3 +166,20 @@ TEST(Runtime, WHEN_partial_or_total_cancellation_alone_is_requested_THEN_delay_r
    EXPECT_FALSE(*result) << result->message();
 }
 #endif
+
+// =================================================================================================
+
+TEST(Runtime, WHEN_timer_is_rearmed_or_cancelled_THEN_only_the_last_callback_runs)
+{
+   IoContext context;
+   Timer timer(context.get_executor());
+   std::vector<int> fired;
+
+   timer.arm(1ms, [&] { fired.push_back(1); });
+   timer.arm(2ms, [&] { fired.push_back(2); });
+   Timer other(context.get_executor());
+   other.arm(1ms, [&] { fired.push_back(3); });
+   other.cancel();
+   context.run();
+   EXPECT_THAT(fired, ElementsAre(2));
+}