Do not treat server as deadlocked when outdated acks are received (#495)

Summary: Do not treat server as deadlocked when outdated acks are
received
Type: Fix
Test Plan: UT/CT, Fullstack
Jira: RDKEMW-18258

Author: skywojciechowskim

media/server/gstplayer/source/tasks/generic/Flush.cpp

@@ -79,6 +79,7 @@ void Flush::execute() const
         m_player.stopPositionReportingAndCheckAudioUnderflowTimer();
         m_context.flushOnPrerollController->waitIfRequired(m_type);
 
+        RIALTO_SERVER_LOG_MIL("Sending flush event for %s source.", common::convertMediaSourceType(m_type));
         // Flush source
         GstEvent *flushStart = m_gstWrapper->gstEventNewFlushStart();
         if (!m_gstWrapper->gstElementSendEvent(source, flushStart))

serverManager/common/source/HealthcheckService.cpp

@@ -65,7 +65,7 @@ void HealthcheckService::onPingSent(int serverId, int pingId)
         return;
     }
     m_remainingPings.insert(serverId);
-    m_failedPings.try_emplace(serverId, 0);
+    m_failedPings.try_emplace(serverId, std::set<int>{});
 }
 
 void HealthcheckService::onPingFailed(int serverId, int pingId)
@@ -86,7 +86,7 @@ void HealthcheckService::onPingFailed(int serverId, int pingId)
     {
         m_sessionServerAppManager.onSessionServerStateChanged(serverId,
                                                               firebolt::rialto::common::SessionServerState::ERROR);
-        m_failedPings.emplace(serverId, 1);
+        m_failedPings.emplace(serverId, std::set<int>{pingId});
     }
 }
 
@@ -95,15 +95,26 @@ void HealthcheckService::onAckReceived(int serverId, int pingId, bool success)
     std::unique_lock<std::mutex> lock{m_mutex};
     if (pingId != m_currentPingId)
     {
-        RIALTO_SERVER_MANAGER_LOG_WARN("Unexpected ack received from server id: %d. Current ping id: %d, received ping "
-                                       "id: %d",
-                                       serverId, m_currentPingId, pingId);
+        if (success && m_failedPings[serverId].find(pingId) != m_failedPings[serverId].end())
+        {
+            RIALTO_SERVER_MANAGER_LOG_WARN("Late ack received for server id: %d, Current ping id: %d, received ping "
+                                           "id: %d. Removing from failed pings list",
+                                           serverId, m_currentPingId, pingId);
+            m_failedPings[serverId].erase(pingId);
+        }
+        else
+        {
+            RIALTO_SERVER_MANAGER_LOG_ERROR("Unexpected ack received from server id: %d. Current ping id: %d, received "
+                                            "ping "
+                                            "id: %d",
+                                            serverId, m_currentPingId, pingId);
+        }
         return;
     }
     m_remainingPings.erase(serverId);
     if (success)
     {
-        m_failedPings[serverId] = 0;
+        m_failedPings[serverId].clear();
     }
     else
     {
@@ -136,12 +147,13 @@ void HealthcheckService::sendPing()
 void HealthcheckService::handleError(int serverId)
 {
     m_sessionServerAppManager.onSessionServerStateChanged(serverId, firebolt::rialto::common::SessionServerState::ERROR);
-    unsigned &failedPingsNum{m_failedPings[serverId]};
-    if (++failedPingsNum >= m_kNumOfFailedPingsBeforeRecovery)
+    auto &failedPings{m_failedPings[serverId]};
+    failedPings.insert(m_currentPingId);
+    if (failedPings.size() >= m_kNumOfFailedPingsBeforeRecovery)
     {
         RIALTO_SERVER_MANAGER_LOG_WARN(
             "Max num of failed pings reached for server with id: %d. Starting recovery action", serverId);
-        failedPingsNum = 0;
+        failedPings.clear();
         m_sessionServerAppManager.restartServer(serverId);
     }
 }

serverManager/common/source/HealthcheckService.h

@@ -53,7 +53,7 @@ class HealthcheckService : public IHealthcheckService
     std::mutex m_mutex;
     int m_currentPingId;
     std::set<int> m_remainingPings;
-    std::map<int, unsigned> m_failedPings;
+    std::map<int, std::set<int>> m_failedPings;
 };
 } // namespace rialto::servermanager::common
 

tests/unittests/serverManager/unittests/common/HealthcheckServiceTests.cpp

@@ -285,3 +285,22 @@ TEST_F(HealthcheckServiceTests, WillFailToFailPingWithWrongId)
     // There should be no error indication here.
     triggerOnPingFailed(pingId + 1);
 }
+
+TEST_F(HealthcheckServiceTests, WillSkipRestartingServerWhenAcksAreReceivedLater)
+{
+    int pingId{-1};
+    timerWillBeCreated();
+    createSut();
+    pingWillBeSent(pingId);
+    triggerPingTimeout();
+    const int firstPingId{pingId};
+    triggerOnPingSent(pingId);
+    errorIndicationWillBeSent();
+    pingWillBeSent(pingId);
+    triggerPingTimeout();
+    triggerOnAckReceived(kServerId, firstPingId, kSuccess);
+    triggerOnPingSent(pingId);
+    errorIndicationWillBeSent();
+    pingWillBeSent(pingId);
+    triggerPingTimeout();
+}