[Driver/C] Fix a race condition when there are only spies and spies simulate connection, i.e. adding/removing spies mutated the array that the sender thread was iterating over. Now conductor computes max spy position while processing stream counters and share it via a volatile int64_t field.

Co-authored-by: Nate Bradac <38217740+nbradac@users.noreply.github.com>
(cherry picked from commit 66a55df)

Author: vyazelenko

aeron-driver/src/main/c/aeron_network_publication.c

@@ -673,7 +673,10 @@ int aeron_network_publication_send(aeron_network_publication_t *publication, int
 
         if (publication->spies_simulate_connection && has_spies && !has_receivers)
         {
-            const int64_t new_snd_pos = aeron_network_publication_max_spy_position(publication, snd_pos);
+            int64_t max_spy_position;
+            AERON_GET_ACQUIRE(max_spy_position, publication->conductor_fields.max_spy_position);
+
+            const int64_t new_snd_pos = max_spy_position > snd_pos ? max_spy_position : snd_pos;
             aeron_counter_set_release(publication->snd_pos_position.value_addr, new_snd_pos);
 
             int64_t flow_control_position = publication->flow_control->on_idle(
@@ -1013,6 +1016,7 @@ int aeron_network_publication_update_pub_pos_and_lmt(aeron_network_publication_t
             int64_t min_consumer_position = snd_pos;
             if (publication->conductor_fields.subscribable.length > 0)
             {
+                int64_t max_consumer_position = snd_pos;
                 for (size_t i = 0, length = publication->conductor_fields.subscribable.length; i < length; i++)
                 {
                     aeron_tetherable_position_t *tetherable_position =
@@ -1022,8 +1026,13 @@ int aeron_network_publication_update_pub_pos_and_lmt(aeron_network_publication_t
                     {
                         int64_t position = aeron_counter_get_acquire(tetherable_position->value_addr);
                         min_consumer_position = position < min_consumer_position ? position : min_consumer_position;
+                        max_consumer_position = position > max_consumer_position ? position : max_consumer_position;
                     }
                 }
+                if (max_consumer_position > publication->conductor_fields.max_spy_position)
+                {
+                    AERON_SET_RELEASE(publication->conductor_fields.max_spy_position, max_consumer_position);
+                }
             }
 
             int64_t new_limit_position = min_consumer_position + publication->term_window_length;
@@ -1438,8 +1447,6 @@ extern void aeron_network_publication_sender_release(aeron_network_publication_t
 
 extern bool aeron_network_publication_has_sender_released(aeron_network_publication_t *publication);
 
-extern int64_t aeron_network_publication_max_spy_position(aeron_network_publication_t *publication, int64_t snd_pos);
-
 extern bool aeron_network_publication_is_accepting_subscriptions(aeron_network_publication_t *publication);
 
 extern inline int64_t aeron_network_publication_registration_id(aeron_network_publication_t *publication);

aeron-driver/src/main/c/aeron_network_publication.h

@@ -50,6 +50,7 @@ typedef struct aeron_network_publication_stct
         int32_t refcnt;
         aeron_driver_managed_resource_t managed_resource;
         aeron_subscribable_t subscribable;
+        volatile int64_t max_spy_position;
         int64_t clean_position;
         int64_t time_of_last_activity_ns;
         int64_t last_snd_pos;
@@ -287,24 +288,6 @@ inline bool aeron_network_publication_has_sender_released(aeron_network_publicat
     return has_sender_released;
 }
 
-inline int64_t aeron_network_publication_max_spy_position(aeron_network_publication_t *publication, int64_t snd_pos)
-{
-    int64_t position = snd_pos;
-
-    for (size_t i = 0, length = publication->conductor_fields.subscribable.length; i < length; i++)
-    {
-        aeron_tetherable_position_t *tetherable_position = &publication->conductor_fields.subscribable.array[i];
-        int64_t spy_position = aeron_counter_get_acquire(tetherable_position->value_addr);
-
-        if (aeron_driver_subscribable_is_active_state(tetherable_position->state))
-        {
-            position = spy_position > position ? spy_position : position;
-        }
-    }
-
-    return position;
-}
-
 inline bool aeron_network_publication_is_accepting_subscriptions(aeron_network_publication_t *publication)
 {
     return AERON_NETWORK_PUBLICATION_STATE_ACTIVE == publication->conductor_fields.state ||

aeron-driver/src/test/c/CMakeLists.txt

@@ -128,6 +128,10 @@ if (AERON_UNIT_TESTS)
     aeron_driver_test(position_test aeron_position_test.cpp)
     aeron_driver_test(driver_context_config_test aeron_driver_context_config_test.cpp)
 
+    aeron_driver_test(c_spy_subscribable_race_test aeron_spy_subscribable_race_test.cpp)
+    set_tests_properties(c_spy_subscribable_race_test PROPERTIES TIMEOUT 60)
+    set_tests_properties(c_spy_subscribable_race_test PROPERTIES RUN_SERIAL TRUE)
+
     if ("${CMAKE_SYSTEM_NAME}" MATCHES "Linux")
         add_executable(aeronmd_signal_test aeronmd_signal_test.cpp)
         add_dependencies(aeronmd_signal_test gmock aeronmd)

aeron-driver/src/test/c/aeron_spy_subscribable_race_test.cpp

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2014-2025 Real Logic Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <atomic>
+#include <thread>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "aeron_test_base.h"
+
+#define SPY_RACE_PUB_URI   "aeron:udp?endpoint=localhost:24370"
+#define SPY_RACE_SPY_URI   "aeron-spy:aeron:udp?endpoint=localhost:24370"
+#define SPY_RACE_STREAM_ID (1234)
+#define SPY_RACE_NUM_SPIES (32)
+
+class SpySubscribableRaceTest : public CSystemTestBase, public testing::Test
+{
+public:
+    SpySubscribableRaceTest() : CSystemTestBase(
+        {},
+        [](aeron_driver_context_t *ctx)
+        {
+            aeron_driver_context_set_threading_mode(ctx, AERON_THREADING_MODE_DEDICATED);
+            aeron_driver_context_set_spies_simulate_connection(ctx, true);
+
+            aeron_driver_context_set_sender_idle_strategy(ctx, "noop");
+            aeron_driver_context_set_conductor_idle_strategy(ctx, "noop");
+            aeron_driver_context_set_receiver_idle_strategy(ctx, "noop");
+
+            aeron_driver_context_set_term_buffer_length(ctx, 64 * 1024);
+        })
+    {}
+};
+
+TEST_F(SpySubscribableRaceTest, shouldNotRaceWhenAddingSpiesWhileSending)
+{
+    ASSERT_TRUE(connect()) << aeron_errmsg();
+
+    aeron_async_add_publication_t *pub_async = nullptr;
+    ASSERT_EQ(aeron_async_add_publication(
+        &pub_async, m_aeron, SPY_RACE_PUB_URI, SPY_RACE_STREAM_ID), 0);
+    aeron_publication_t *publication = awaitPublicationOrError(pub_async);
+    ASSERT_TRUE(publication) << aeron_errmsg();
+
+    std::atomic<bool> stop_sender{ false };
+    std::thread sender_thread(
+        [&]()
+        {
+            static const uint8_t payload[32] = {};
+            while (!stop_sender.load(std::memory_order_relaxed))
+            {
+                aeron_publication_offer(publication, payload, sizeof(payload), nullptr, nullptr);
+            }
+        });
+
+    std::vector<aeron_async_add_subscription_t *> sub_asyncs(SPY_RACE_NUM_SPIES, nullptr);
+    std::vector<aeron_subscription_t *>           subs(SPY_RACE_NUM_SPIES, nullptr);
+
+    for (int i = 0; i < SPY_RACE_NUM_SPIES; i++)
+    {
+        ASSERT_EQ(aeron_async_add_subscription(
+            &sub_asyncs[i], m_aeron, SPY_RACE_SPY_URI, SPY_RACE_STREAM_ID,
+            nullptr, nullptr, nullptr, nullptr), 0)
+            << "spy " << i << ": " << aeron_errmsg();
+    }
+
+    for (int i = 0; i < SPY_RACE_NUM_SPIES; i++)
+    {
+        subs[i] = awaitSubscriptionOrError(sub_asyncs[i]);
+        ASSERT_TRUE(subs[i]) << "spy " << i << " registration failed: " << aeron_errmsg();
+    }
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    std::atomic<int> closes_remaining{ SPY_RACE_NUM_SPIES };
+    auto on_close = [](void *clientd)
+    {
+        static_cast<std::atomic<int> *>(clientd)->fetch_sub(1, std::memory_order_release);
+    };
+
+    for (auto *sub : subs)
+    {
+        aeron_subscription_close(sub, on_close, &closes_remaining);
+    }
+
+    while (closes_remaining.load(std::memory_order_acquire) > 0)
+    {
+        std::this_thread::yield();
+    }
+
+    stop_sender.store(true);
+    sender_thread.join();
+
+    aeron_publication_close(publication, nullptr, nullptr);
+}