diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index e55ed40607..e285152455 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1470,10 +1470,7 @@ void gossiper::update_timestamp_for_nodes(const std::map<inet_address, endpoint_
 }
 
 void gossiper::mark_alive(inet_address addr, endpoint_state& local_state) {
-    // if (MessagingService.instance().getVersion(addr) < MessagingService.VERSION_20) {
-    //     real_mark_alive(addr, local_state);
-    //     return;
-    // }
+    // Enter the _background_msg gate so stop() would wait on it
     auto inserted = _pending_mark_alive_endpoints.insert(addr).second;
     if (inserted) {
         // The node is not in the _pending_mark_alive_endpoints
@@ -1484,6 +1481,11 @@ void gossiper::mark_alive(inet_address addr, endpoint_state& local_state) {
         return;
     }
 
+    // unmark addr as pending on exception or after background continuation completes
+    auto unmark_pending = deferred_action([this, addr, g = shared_from_this()] () noexcept {
+        _pending_mark_alive_endpoints.erase(addr);
+    });
+
     local_state.mark_dead();
     msg_addr id = get_msg_addr(addr);
     auto generation = _endpoint_state_map[get_broadcast_address()].get_heart_beat_state().get_generation();
@@ -1504,9 +1506,7 @@ void gossiper::mark_alive(inet_address addr, endpoint_state& local_state) {
             return real_mark_alive(addr, state);
         }
         return make_ready_future();
-    }).finally([this, addr] {
-        _pending_mark_alive_endpoints.erase(addr);
-    }).handle_exception([addr, gh = std::move(gh)] (auto ep) {
+    }).handle_exception([addr, gh = std::move(gh), unmark_pending = std::move(unmark_pending)] (auto ep) {
         logger.warn("Fail to send EchoMessage to {}: {}", addr, ep);
     });
 }