From 1e753952092d0257a516749bb598262d94ae0431 Mon Sep 17 00:00:00 2001
From: Povilas Kanapickas <povilas@radix.lt>
Date: Wed, 15 Jan 2025 17:46:56 +0200
Subject: [PATCH] Revert "rpc stream: do not abort stream queue if stream
 connection was closed without error"

This reverts commit 27f834e8f04f6b7ff9521424e88f0f32e5d53dc1.

This revert fixes test failures in Seastar.unit.rpc test suite, such as:

      Start 47: Seastar.unit.rpc
47/79 Test #47: Seastar.unit.rpc ..............................***Timeout 300.04 sec

The tests fail randomly and more frequently in debug builds.

Closes scylladb/seastar#2618
---
 src/rpc/rpc.cc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/rpc/rpc.cc b/src/rpc/rpc.cc
index a360535812..88d2660c52 100644
--- a/src/rpc/rpc.cc
+++ b/src/rpc/rpc.cc
@@ -1021,9 +1021,9 @@ namespace rpc {
                       log_exception(*this, log_level::debug, "fail to connect", ep);
                   }
               }
-            _stream_queue.abort(ep);
           }
           _error = true;
+          _stream_queue.abort(std::make_exception_ptr(stream_closed()));
           return stop_send_loop(ep).then_wrapped([this] (future<> f) {
               f.ignore_ready_future();
               _outstanding.clear();
@@ -1242,10 +1242,10 @@ future<> server::connection::send_unknown_verb_reply(std::optional<rpc_clock_typ
               ep = f.get_exception();
               log_exception(*this, log_level::error,
                       format("server{} connection dropped", is_stream() ? " stream" : "").c_str(), ep);
-            _stream_queue.abort(ep);
           }
           _fd.shutdown_input();
           _error = true;
+          _stream_queue.abort(std::make_exception_ptr(stream_closed()));
           return stop_send_loop(ep).then_wrapped([this] (future<> f) {
               f.ignore_ready_future();
               get_server()._conns.erase(get_connection_id());