Harden rDSN core, RPC, and tooling against crashes on recoverable errors

include/dsn/tool-api/task.h

@@ -307,6 +307,28 @@ class rpc_request_task : public task, public transient_object
             static_cast<uint64_t>(_request->header->client.timeout_ms) * 1000000ULL)
         {
             _handler->run(_request);
+            // rpc_handler_info::run() consumes the dispatch reference.
+            _handler = nullptr;
+        }
+        else
+        {
+            release_handler();
+        }
+    }
+
+private:
+    void release_handler()
+    {
+        rpc_handler_info* handler = _handler;
+        if (handler == nullptr)
+        {
+            return;
+        }
+
+        _handler = nullptr;
+        if (1 == handler->release_ref())
+        {
+            delete handler;
         }
     }
 

src/core/src/env_provider.cpp

@@ -65,7 +65,17 @@ void env_provider::set_thread_local_random_seed(int s)
 
 uint64_t env_provider::random64(uint64_t min, uint64_t max)
 {
-    dassert(min <= max, "invalid random range");
+    if (min > max)
+    {
+        // Callers may derive min/max from independent configuration values (e.g. the
+        // emulator's min/max message delay). An inverted range aborts here via the
+        // assert and, once asserts are compiled out, feeds an invalid range to
+        // uniform_int_distribution (undefined behavior). Degrade to a deterministic
+        // in-range value instead of crashing.
+        dwarn("invalid random range [%llu, %llu], returning %llu",
+              (unsigned long long)min, (unsigned long long)max, (unsigned long long)min);
+        return min;
+    }
     if (env_provider__tls_magic != 0xdeadbeef)
     {
         env_provider__rng = new std::remove_pointer<decltype(env_provider__rng)>::type(std::random_device{}());

src/core/src/group_address.h

@@ -131,7 +131,23 @@ namespace dsn
 
     inline rpc_address rpc_group_address::possible_leader()
     {
-        alr_t l(_lock);
+        {
+            // Fast path: a leader has already been chosen, so a shared read lock
+            // is enough here (we only read _members/_leader_index).
+            alr_t l(_lock);
+            if (_members.empty())
+                return _invalid;
+            if (_leader_index != -1)
+                return _members[_leader_index];
+        }
+
+        // Slow path: no leader yet, so we must assign _leader_index, which requires
+        // the exclusive write lock (writing it under the shared read lock would be a
+        // data race; the sibling writers leader_forward()/set_leader() also take the
+        // write lock). Re-check under the write lock because another thread may have
+        // chosen the leader (or emptied _members) between releasing the read lock and
+        // acquiring the write lock.
+        alw_t l(_lock);
         if (_members.empty())
             return _invalid;
         if (_leader_index == -1)

src/core/src/main.cpp

@@ -602,7 +602,11 @@ bool run(
     for (auto it = spec.toollets.begin(); it != spec.toollets.end(); ++it)
     {
         auto tlet = dsn::tools::internal_use_only::get_toollet(it->c_str(), ::dsn::PROVIDER_TYPE_MAIN);
-        dassert(tlet, "toolet not found");
+        if (nullptr == tlet)
+        {
+            derror("toollet '%s' is not found, please check the [core] toollets configuration", it->c_str());
+            return false;
+        }
         tlet->install(spec);
     }
 
@@ -612,7 +616,12 @@ bool run(
     // TODO: register sys_exit execution
 
     // init runtime
-    ::dsn::service_engine::fast_instance().init_after_toollets();
+    auto err = ::dsn::service_engine::fast_instance().init_after_toollets();
+    if (err != ::dsn::ERR_OK)
+    {
+        derror("service engine init failed, err = %s", err.to_string());
+        return false;
+    }
 
     dsn_all.engine_ready.store(true, std::memory_order_release);
 
@@ -638,6 +647,14 @@ bool run(
             {
                 ::dsn::safe_list< ::dsn::safe_string> argskvs;
                 ::dsn::utils::split_args(kv.c_str(), argskvs, '@');
+                // split_args drops empty/whitespace-only tokens, so a malformed app_list
+                // entry consisting only of '@' separators (e.g. -app_list "@") yields an
+                // empty list. front() on an empty std::list is undefined behavior, so skip
+                // such an entry: it names no app and cannot match any config section.
+                if (argskvs.empty())
+                {
+                    continue;
+                }
                 if (::dsn::safe_string("apps.") + argskvs.front() == sp.config_section)
                 {
                     if (argskvs.size() < 2)

src/core/src/network.cpp

@@ -579,7 +579,8 @@ namespace dsn
             char name[128];
             if (gethostname(name, sizeof(name)) != 0)
             {
-                dassert(false, "gethostname failed, err = %s", strerror(errno));
+                derror("gethostname failed, err = %s", strerror(errno));
+                return 0;
             }
             ip = dsn_ipv4_from_host(name);
         }

src/core/src/rpc.test.cpp

@@ -37,6 +37,7 @@
 #include <dsn/service_api_cpp.h>
 #include <dsn/utility/priority_queue.h>
 #include "group_address.h"
+#include "rpc_engine.h"
 #include <dsn/cpp/test_utils.h>
 #include <vector>
 #include <string>
@@ -45,6 +46,23 @@
 
 typedef std::function<void(error_code, dsn_message_t, dsn_message_t)> rpc_reply_handler;
 
+namespace {
+
+class expired_rpc_request_task : public ::dsn::rpc_request_task
+{
+public:
+    expired_rpc_request_task(::dsn::message_ex* request,
+                             ::dsn::rpc_handler_info* handler,
+                             ::dsn::service_node* node)
+        : rpc_request_task(request, handler, node)
+    {
+    }
+
+    void expire() { _enqueue_ts_ns = 1; }
+};
+
+} // anonymous namespace
+
 static ::dsn::rpc_address build_group() {
     ::dsn::rpc_address server_group;
     server_group.assign_group(dsn_group_build("server_group.test"));
@@ -93,6 +111,82 @@ TEST(core, rpc)
     EXPECT_TRUE(result.second == "server");
 }
 
+TEST(core, rpc_inline_dispatch_releases_handler_reference)
+{
+    ::dsn::rpc_server_dispatcher dispatcher;
+    ::dsn::rpc_handler_info handler(RPC_TEST_HASH);
+    int call_count = 0;
+
+    handler.name = "rpc.inline.reference";
+    handler.c_handler = [](dsn_message_t, void* context) {
+        ++*static_cast<int*>(context);
+    };
+    handler.parameter = &call_count;
+    handler.add_ref();
+    ASSERT_TRUE(dispatcher.register_rpc_handler(&handler));
+
+    auto request = ::dsn::message_ex::create_request(RPC_TEST_HASH, 100, 0, 0);
+    ASSERT_NE(nullptr, request);
+    request->add_ref();
+
+    dispatcher.on_request_with_inline_execution(request, nullptr);
+
+    EXPECT_EQ(1, call_count);
+    EXPECT_EQ(1, handler.running_count.load(std::memory_order_relaxed));
+    EXPECT_EQ(&handler, dispatcher.unregister_rpc_handler(RPC_TEST_HASH));
+
+    while (handler.running_count.load(std::memory_order_relaxed) > 0)
+    {
+        handler.release_ref();
+    }
+    request->release_ref();
+}
+
+TEST(core, rpc_expired_dispatch_releases_handler_reference)
+{
+    ::dsn::rpc_handler_info handler(RPC_TEST_HASH);
+    int call_count = 0;
+
+    handler.c_handler = [](dsn_message_t, void* context) {
+        ++*static_cast<int*>(context);
+    };
+    handler.parameter = &call_count;
+    handler.add_ref();
+    handler.add_ref();
+
+    auto request = ::dsn::message_ex::create_request(RPC_TEST_HASH, 0, 0, 0);
+    ASSERT_NE(nullptr, request);
+    {
+        expired_rpc_request_task task(request, &handler, nullptr);
+        task.expire();
+        task.exec();
+    }
+
+    EXPECT_EQ(0, call_count);
+    EXPECT_EQ(1, handler.running_count.load(std::memory_order_relaxed));
+    while (handler.running_count.load(std::memory_order_relaxed) > 0)
+    {
+        handler.release_ref();
+    }
+}
+
+TEST(core, rpc_discarded_dispatch_releases_handler_reference)
+{
+    ::dsn::rpc_handler_info handler(RPC_TEST_HASH);
+    handler.add_ref();
+    handler.add_ref();
+
+    auto request = ::dsn::message_ex::create_request(RPC_TEST_HASH, 100, 0, 0);
+    ASSERT_NE(nullptr, request);
+    { ::dsn::rpc_request_task task(request, &handler, nullptr); }
+
+    EXPECT_EQ(1, handler.running_count.load(std::memory_order_relaxed));
+    while (handler.running_count.load(std::memory_order_relaxed) > 0)
+    {
+        handler.release_ref();
+    }
+}
+
 TEST(core, group_address_talk_to_others)
 {
     ::dsn::rpc_address addr = build_group();

src/core/src/rpc_engine.cpp

@@ -455,7 +455,7 @@ namespace dsn {
         }
         else
         {
-            dassert(false, "rpc registration confliction for '%s'", name.c_str());
+            derror("rpc registration confliction for '%s'", name.c_str());
             return false;
         }
     }
@@ -547,6 +547,10 @@ namespace dsn {
         if (handler)
         {
             handler->c_handler(msg, handler->parameter);
+            if (1 == handler->release_ref())
+            {
+                delete handler;
+            }
         }
         else
         {
@@ -636,13 +640,26 @@ namespace dsn {
         const service_spec& spec = service_engine::fast_instance().spec();
         network* net = utils::factory_store<network>::create(
             netcs.factory_name.c_str(), ::dsn::PROVIDER_TYPE_MAIN, this, nullptr);
+        if (net == nullptr)
+        {
+            derror("cannot create network provider '%s', please check the factory registration and configuration",
+                netcs.factory_name.c_str());
+            return nullptr;
+        }
         net->reset_parser_attr(client_hdr_format, netcs.message_buffer_block_size);
 
         for (auto it = spec.network_aspects.begin();
             it != spec.network_aspects.end();
             it++)
         {
-            net = utils::factory_store<network>::create(it->c_str(), ::dsn::PROVIDER_TYPE_ASPECT, this, net);
+            network* net2 = utils::factory_store<network>::create(it->c_str(), ::dsn::PROVIDER_TYPE_ASPECT, this, net);
+            if (net2 == nullptr)
+            {
+                derror("cannot create network aspect provider '%s', please check the factory registration and configuration",
+                    it->c_str());
+                return nullptr;
+            }
+            net = net2;
         }
 
         // start the net
@@ -939,7 +956,18 @@ namespace dsn {
 
                                 // still got time, retry
                                 uint64_t nms = dsn_now_ms();
-                                uint64_t gap = 8 << req2->send_retry_count;
+                                // Bound the shift width: 8 << 7 (1024) already exceeds the
+                                // 1000ms ceiling below, so clamping the exponent leaves the
+                                // exponential backoff identical while avoiding the signed
+                                // overflow / oversized-shift undefined behavior that occurs
+                                // once send_retry_count grows large during long-timeout
+                                // retries against a persistently failing partition.
+                                int retry_shift = req2->send_retry_count;
+                                if (retry_shift > 10)
+                                {
+                                    retry_shift = 10;
+                                }
+                                uint64_t gap = 8ull << retry_shift;
                                 if (gap > 1000)
                                     gap = 1000;
                                 if (nms + gap < timeout_ts_ms)

src/core/src/service_api_c.test.cpp

@@ -526,6 +526,24 @@ TEST(core, dsn_rpc_registration_invalid_parameters)
     ASSERT_EQ(nullptr, dsn_rpc_unregiser_handler(TASK_CODE_INVALID, dsn_gpid()));
 }
 
+TEST(core, dsn_rpc_registration_conflict)
+{
+    int first_context = 1;
+    int second_context = 2;
+
+    ASSERT_TRUE(dsn_rpc_register_handler(TASK_CODE_RPC_FOR_TEST,
+                                         "first_handler",
+                                         noop_rpc_request_handler,
+                                         &first_context,
+                                         dsn_gpid()));
+    ASSERT_FALSE(dsn_rpc_register_handler(TASK_CODE_RPC_FOR_TEST,
+                                          "second_handler",
+                                          noop_rpc_request_handler,
+                                          &second_context,
+                                          dsn_gpid()));
+    ASSERT_EQ(&first_context, dsn_rpc_unregiser_handler(TASK_CODE_RPC_FOR_TEST, dsn_gpid()));
+}
+
 TEST(core, dsn_rpc_dispatch_invalid_parameters)
 {
     const dsn_address_t invalid_address = {};

src/core/src/service_engine.cpp

@@ -201,6 +201,12 @@ error_code service_node::init_io_engine(io_engine& io, ioe_mode mode)
             aio = factory_store<aio_provider>::create(it->c_str(),
                 PROVIDER_TYPE_ASPECT, io.disk, aio);
         }
+        if (nullptr == aio)
+        {
+            derror("cannot create aio provider '%s', please check [core] aio_factory_name and aio_aspects",
+                spec.aio_factory_name.c_str());
+            return ERR_SERVICE_NOT_FOUND;
+        }
         io.aio = aio;
     }
     else
@@ -580,7 +586,7 @@ void service_engine::init_before_toollets(const service_spec& spec)
         );
 }
 
-void service_engine::init_after_toollets()
+error_code service_engine::init_after_toollets()
 {
     // init common providers (second half)
     _env = factory_store<env_provider>::create(_spec.env_factory_name.c_str(), 
@@ -592,7 +598,14 @@ void service_engine::init_after_toollets()
         _env = factory_store<env_provider>::create(it->c_str(), 
             PROVIDER_TYPE_ASPECT, _env);
     }
+    if (nullptr == _env)
+    {
+        derror("cannot create env provider '%s', please check [core] env_factory_name and env_aspects",
+            _spec.env_factory_name.c_str());
+        return ERR_SERVICE_NOT_FOUND;
+    }
     tls_dsn.env = _env;
+    return ERR_OK;
 }
 
 void service_engine::register_system_rpc_handler(
@@ -666,12 +679,17 @@ service_node* service_engine::start_node(service_app_spec& app_spec)
             {
                 service_node* n = _nodes_by_app_port[p];
 
-                dassert(false, "network port %d usage confliction for %s vs %s, "
+                // A port collision between two configured apps is an operator
+                // misconfiguration, not an internal invariant violation. Report it and
+                // fail node creation gracefully (callers below already handle a nullptr
+                // return) instead of aborting the whole process.
+                derror("network port %d usage confliction for %s vs %s, "
                     "please reconfig",
                     p,
                     n->name(),
                     app_spec.name.c_str()
                     );
+                return nullptr;
             }
         }
 

src/core/src/service_engine.h

@@ -155,7 +155,7 @@ class service_engine : public utils::singleton<service_engine>
     static safe_string get_queue_info(const safe_vector<safe_string>& args);
 
     void init_before_toollets(const service_spec& spec);
-    void init_after_toollets();
+    error_code init_after_toollets();
     void configuration_changed();
 
     service_node* start_node(service_app_spec& app_spec);