Fix stack-use-after-return in client_routes delete/set functions

Change lambda captures in set_client_routes and delete_client_routes to move
route_keys/route_entries into the inner lambda instead of capturing by reference.
This prevents use-after-return when the with_retry coroutine suspends.

Co-authored-by: mykaul <4655593+mykaul@users.noreply.github.com>

service/client_routes.cc

@@ -105,7 +105,7 @@ seastar::future<> service::client_routes_service::delete_client_routes_inner(con
 
 seastar::future<> service::client_routes_service::set_client_routes(const std::vector<service::client_routes_service::client_route_entry>& route_entries) {
     return container().invoke_on(0, [route_entries = std::move(route_entries)] (service::client_routes_service& cr) -> future<> {
-        return cr.with_retry([&] {
+        return cr.with_retry([&cr, route_entries = std::move(route_entries)] () mutable {
             return cr.set_client_routes_inner(route_entries);
         });
     });
@@ -113,7 +113,7 @@ seastar::future<> service::client_routes_service::set_client_routes(const std::v
 
 seastar::future<> service::client_routes_service::delete_client_routes(const std::vector<service::client_routes_service::client_route_key>& route_keys) {
     return container().invoke_on(0, [route_keys = std::move(route_keys)] (service::client_routes_service& cr) -> future<> {
-        return cr.with_retry([&] {
+        return cr.with_retry([&cr, route_keys = std::move(route_keys)] () mutable {
             return cr.delete_client_routes_inner(route_keys);
         });
     });

utils/atomic_vector.hh

@@ -15,8 +15,8 @@
 
 #include <vector>
 
-// This class supports atomic inserts, removes, and iteration.
-// All operations are synchronized using a read-write lock.
+// This class supports atomic removes (by using a lock and returning a
+// future) and non atomic insert and iteration (by using indexes).
 template <typename T>
 class atomic_vector {
     std::vector<T> _vec;
@@ -24,10 +24,6 @@ class atomic_vector {
 
 public:
     void add(const T& value) {
-        auto lock = _vec_lock.for_write().lock().get();
-        auto unlock = seastar::defer([this] {
-            _vec_lock.for_write().unlock();
-        });
         _vec.push_back(value);
     }
     seastar::future<> remove(const T& value) {
@@ -48,14 +44,11 @@ public:
         auto unlock = seastar::defer([this] {
             _vec_lock.for_read().unlock();
         });
-        // Take a snapshot of the current contents while holding the read lock,
-        // so that concurrent add() calls and possible reallocations won't
-        // affect our iteration.
-        auto snapshot = _vec;
-        // We grab locks in both add() and remove(), so we iterate using
-        // indexes on the snapshot to avoid concurrent modifications.
-        for (size_t i = 0, n = snapshot.size(); i < n; ++i) {
-            func(snapshot[i]);
+        // We grab a lock in remove(), but not in add(), so we
+        // iterate using indexes to guard against the vector being
+        // reallocated.
+        for (size_t i = 0, n = _vec.size(); i < n; ++i) {
+            func(_vec[i]);
         }
     }
 
@@ -66,17 +59,16 @@ public:
     void thread_for_each_nested(seastar::noncopyable_function<void(T)> func) const {
         // When called in the context of thread_for_each, the read lock is
         // already held, so we don't need to acquire it again. Acquiring it
-        // again could lead to a deadlock. This function must only be called
-        // while holding the read lock on _vec_lock.
+        // again could lead to a deadlock.
+        if (!_vec_lock.locked()) {
+            utils::on_internal_error("thread_for_each_nested called without holding the vector lock");
+        }
 
-        // Take a snapshot of the current contents while the read lock is held,
-        // so that concurrent add() calls and possible reallocations won't
-        // affect our iteration.
-        auto snapshot = _vec;
-        // We grab locks in both add() and remove(), so we iterate using
-        // indexes on the snapshot to avoid concurrent modifications.
-        for (size_t i = 0, n = snapshot.size(); i < n; ++i) {
-            func(snapshot[i]);
+        // We grab a lock in remove(), but not in add(), so we
+        // iterate using indexes to guard against the vector being
+        // reallocated.
+        for (size_t i = 0, n = _vec.size(); i < n; ++i) {
+            func(_vec[i]);
         }
     }
 
@@ -87,14 +79,11 @@ public:
     // preemption.
     seastar::future<> for_each(seastar::noncopyable_function<seastar::future<>(T)> func) const {
         auto holder = co_await _vec_lock.hold_read_lock();
-        // Take a snapshot of the current contents while holding the read lock,
-        // so that concurrent add() calls and possible reallocations won't
-        // affect our iteration.
-        auto snapshot = _vec;
-        // We grab locks in both add() and remove(), so we iterate using
-        // indexes on the snapshot to avoid concurrent modifications.
-        for (size_t i = 0, n = snapshot.size(); i < n; ++i) {
-            co_await func(snapshot[i]);
+        // We grab a lock in remove(), but not in add(), so we
+        // iterate using indexes to guard against the vector being
+        // reallocated.
+        for (size_t i = 0, n = _vec.size(); i < n; ++i) {
+            co_await func(_vec[i]);
         }
     }
 };