Integration fixes

.github/workflows/build-test.yaml

@@ -41,6 +41,10 @@ jobs:
           repository: ${{github.event.pull_request.head.repo.full_name}}
           fetch-depth: 0
 
+      - name: Static analysis
+        run: |
+          ./build.sh lint --aos-service cm --aos-service iam --aos-service sm
+
       - name: Install SonarQube build wrapper
         uses: SonarSource/sonarqube-scan-action/install-build-wrapper@v6
 
@@ -56,10 +60,6 @@ jobs:
         run: |
           ./build.sh coverage --aos-service cm --aos-service iam --aos-service sm
 
-      - name: Static analysis
-        run: |
-          ./build.sh lint --aos-service cm --aos-service iam --aos-service sm
-
       - name: Upload codecov report
         uses: codecov/codecov-action@v4
         with:

src/cm/config/config.cpp

@@ -28,7 +28,7 @@ constexpr auto cDefaultLauncherInstanceTTL            = "30d";
 constexpr auto cDefaultLauncherNodesConnectionTimeout = "10m";
 constexpr auto cDefaultMonitoringSendPeriod           = "1m";
 constexpr auto cDefaultSMConnectionTimeout            = "1m";
-constexpr auto cDefaultUnitStatusSendTimeout          = "30s";
+constexpr auto cDefaultUnitStatusSendTimeout          = "10s";
 constexpr auto cDefaultUpdateItemTTL                  = "30d";
 constexpr auto cDefaultRemoveOutdatedPeriod           = "24h";
 constexpr auto cDefaultMigrationPath                  = "/usr/share/aos/communicationmanager/migration";

src/cm/config/tests/config.cpp

@@ -168,7 +168,7 @@ TEST_F(CMConfigTest, ParseMinimalConfigWithDefaults)
     EXPECT_EQ(config.mStateDir, (std::filesystem::path("workingDir") / "states").string());
     EXPECT_EQ(config.mUnitConfigFile, (std::filesystem::path("workingDir") / "aos_unit.cfg").string());
 
-    EXPECT_EQ(config.mUnitStatusSendTimeout, aos::Time::cSeconds * 30);
+    EXPECT_EQ(config.mUnitStatusSendTimeout, aos::Time::cSeconds * 10);
     EXPECT_EQ(config.mCloudResponseWaitTimeout, aos::Time::cSeconds * 10);
 
     EXPECT_EQ(config.mMonitoring.mSendPeriod, aos::Time::cMinutes * 1);

src/cm/database/database.cpp

@@ -656,8 +656,8 @@ Error Database::AddItem(const imagemanager::ItemInfo& item)
         ImageManagerItemInfoRow row;
 
         FromAos(item, row);
-        *mSession
-            << "INSERT INTO imagemanager (itemID, version, indexDigest, state, timestamp) VALUES (?, ?, ?, ?, ?);",
+        *mSession << "INSERT INTO imagemanager (itemID, type, version, indexDigest, state, timestamp) VALUES "
+                     "(?, ?, ?, ?, ?, ?);",
             bind(row), now;
     } catch (const std::exception& e) {
         return AOS_ERROR_WRAP(common::utils::ToAosError(e));
@@ -713,7 +713,7 @@ Error Database::GetAllItemsInfos(Array<imagemanager::ItemInfo>& items)
     try {
         std::vector<ImageManagerItemInfoRow> rows;
 
-        *mSession << "SELECT itemID, version, indexDigest, state, timestamp FROM imagemanager;", into(rows), now;
+        *mSession << "SELECT itemID, type, version, indexDigest, state, timestamp FROM imagemanager;", into(rows), now;
 
         auto itemInfo = std::make_unique<imagemanager::ItemInfo>();
 
@@ -737,7 +737,7 @@ Error Database::GetItemInfos(const String& id, Array<imagemanager::ItemInfo>& it
     try {
         std::vector<ImageManagerItemInfoRow> rows;
 
-        *mSession << "SELECT itemID, version, indexDigest, state, timestamp FROM imagemanager WHERE itemID = ?;",
+        *mSession << "SELECT itemID, type, version, indexDigest, state, timestamp FROM imagemanager WHERE itemID = ?;",
             bind(id.CStr()), into(rows), now;
 
         auto itemInfo = std::make_unique<imagemanager::ItemInfo>();
@@ -874,6 +874,7 @@ void Database::CreateTables()
 
     *mSession << "CREATE TABLE IF NOT EXISTS imagemanager ("
                  "itemID TEXT,"
+                 "type TEXT,"
                  "version TEXT,"
                  "indexDigest TEXT,"
                  "state TEXT,"
@@ -1093,17 +1094,20 @@ void Database::FromAos(const imagemanager::ItemInfo& src, ImageManagerItemInfoRo
 {
     dst.set<ToInt(ImageManagerItemInfoColumns::eItemID)>(src.mItemID.CStr());
     dst.set<ToInt(ImageManagerItemInfoColumns::eVersion)>(src.mVersion.CStr());
+    dst.set<ToInt(ImageManagerItemInfoColumns::eType)>(src.mType.ToString().CStr());
     dst.set<ToInt(ImageManagerItemInfoColumns::eIndexDigest)>(src.mIndexDigest.CStr());
     dst.set<ToInt(ImageManagerItemInfoColumns::eState)>(src.mState.ToString().CStr());
     dst.set<ToInt(ImageManagerItemInfoColumns::eTimestamp)>(src.mTimestamp.UnixNano());
 }
 
 void Database::ToAos(const ImageManagerItemInfoRow& src, imagemanager::ItemInfo& dst)
 {
-    dst.mItemID      = src.get<ToInt(ImageManagerItemInfoColumns::eItemID)>().c_str();
+    dst.mItemID = src.get<ToInt(ImageManagerItemInfoColumns::eItemID)>().c_str();
+    auto err    = dst.mType.FromString(src.get<ToInt(ImageManagerItemInfoColumns::eType)>().c_str());
+    AOS_ERROR_CHECK_AND_THROW(err, "failed to parse item type");
     dst.mVersion     = src.get<ToInt(ImageManagerItemInfoColumns::eVersion)>().c_str();
     dst.mIndexDigest = src.get<ToInt(ImageManagerItemInfoColumns::eIndexDigest)>().c_str();
-    auto err         = dst.mState.FromString(src.get<ToInt(ImageManagerItemInfoColumns::eState)>().c_str());
+    err              = dst.mState.FromString(src.get<ToInt(ImageManagerItemInfoColumns::eState)>().c_str());
     AOS_ERROR_CHECK_AND_THROW(err, "failed to parse item state");
     auto timestamp = src.get<ToInt(ImageManagerItemInfoColumns::eTimestamp)>();
     dst.mTimestamp = Time::Unix(timestamp / Time::cSeconds.Nanoseconds(), timestamp % Time::cSeconds.Nanoseconds());

src/cm/database/database.hpp

@@ -276,6 +276,10 @@ class Database : public DatabaseItf, public networkmanager::StorageItf {
      */
     Error StoreDesiredStatus(const DesiredStatus& desiredStatus) override;
 
+    //
+    // updatemanager::StorageItf interface
+    //
+
     /**
      * Stores update state in storage.
      *
@@ -362,8 +366,16 @@ class Database : public DatabaseItf, public networkmanager::StorageItf {
         std::string, std::string, std::string, uint32_t, uint32_t, uint64_t, std::string, bool, std::string,
         std::string, std::string, std::string, size_t>;
 
-    enum class ImageManagerItemInfoColumns : int { eItemID = 0, eVersion, eIndexDigest, eState, eTimestamp };
-    using ImageManagerItemInfoRow = Poco::Tuple<std::string, std::string, std::string, std::string, uint64_t>;
+    enum class ImageManagerItemInfoColumns : int {
+        eItemID = 0,
+        eType,
+        eVersion,
+        eIndexDigest,
+        eState,
+        eTimestamp,
+    };
+    using ImageManagerItemInfoRow
+        = Poco::Tuple<std::string, std::string, std::string, std::string, std::string, uint64_t>;
 
     // make virtual for unit tests
     virtual int GetVersion() const;

src/cm/database/tests/database.cpp

@@ -136,11 +136,12 @@ launcher::InstanceInfo CreateLauncherInstanceInfo(const char* itemID, const char
 }
 
 imagemanager::ItemInfo CreateImageManagerItemInfo(
-    const char* itemID, const char* version, const char* indexDigest, ItemState state)
+    const char* itemID, const UpdateItemType& type, const char* version, const char* indexDigest, ItemState state)
 {
     imagemanager::ItemInfo info;
 
     info.mItemID      = itemID;
+    info.mType        = type;
     info.mVersion     = version;
     info.mIndexDigest = indexDigest;
     info.mState       = state;
@@ -677,15 +678,19 @@ TEST_F(CMDatabaseTest, ImageManagerAddItem)
 {
     ASSERT_TRUE(mDB.Init(mDatabaseConfig).IsNone());
 
-    auto item1 = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
-    auto item2 = CreateImageManagerItemInfo("service1", "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
-    auto item3 = CreateImageManagerItemInfo("service2", "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
+    auto item1 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
+    auto item2 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
+    auto item3 = CreateImageManagerItemInfo(
+        "service2", UpdateItemTypeEnum::eService, "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
 
     ASSERT_TRUE(mDB.AddItem(item1).IsNone());
     ASSERT_TRUE(mDB.AddItem(item2).IsNone());
     ASSERT_TRUE(mDB.AddItem(item3).IsNone());
 
-    auto duplicateItem = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:xyz999", ItemStateEnum::eInstalled);
+    auto duplicateItem = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:xyz999", ItemStateEnum::eInstalled);
     ASSERT_FALSE(mDB.AddItem(duplicateItem).IsNone());
 
     StaticArray<imagemanager::ItemInfo, 3> items;
@@ -699,9 +704,12 @@ TEST_F(CMDatabaseTest, ImageManagerRemoveItem)
 {
     ASSERT_TRUE(mDB.Init(mDatabaseConfig).IsNone());
 
-    auto item1 = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
-    auto item2 = CreateImageManagerItemInfo("service1", "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
-    auto item3 = CreateImageManagerItemInfo("service2", "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
+    auto item1 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
+    auto item2 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
+    auto item3 = CreateImageManagerItemInfo(
+        "service2", UpdateItemTypeEnum::eService, "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
 
     ASSERT_TRUE(mDB.AddItem(item1).IsNone());
     ASSERT_TRUE(mDB.AddItem(item2).IsNone());
@@ -722,8 +730,10 @@ TEST_F(CMDatabaseTest, ImageManagerUpdateItemState)
 {
     ASSERT_TRUE(mDB.Init(mDatabaseConfig).IsNone());
 
-    auto item1 = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:abc123", ItemStateEnum::ePending);
-    auto item2 = CreateImageManagerItemInfo("service2", "1.0.0", "sha256:def456", ItemStateEnum::ePending);
+    auto item1 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:abc123", ItemStateEnum::ePending);
+    auto item2 = CreateImageManagerItemInfo(
+        "service2", UpdateItemTypeEnum::eService, "1.0.0", "sha256:def456", ItemStateEnum::ePending);
 
     ASSERT_TRUE(mDB.AddItem(item1).IsNone());
     ASSERT_TRUE(mDB.AddItem(item2).IsNone());
@@ -756,9 +766,12 @@ TEST_F(CMDatabaseTest, ImageManagerGetItemsInfo)
     ASSERT_TRUE(mDB.GetAllItemsInfos(emptyItems).IsNone());
     EXPECT_EQ(emptyItems.Size(), 0);
 
-    auto item1 = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
-    auto item2 = CreateImageManagerItemInfo("service1", "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
-    auto item3 = CreateImageManagerItemInfo("service2", "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
+    auto item1 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
+    auto item2 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
+    auto item3 = CreateImageManagerItemInfo(
+        "service2", UpdateItemTypeEnum::eService, "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
 
     ASSERT_TRUE(mDB.AddItem(item1).IsNone());
     ASSERT_TRUE(mDB.AddItem(item2).IsNone());
@@ -775,9 +788,12 @@ TEST_F(CMDatabaseTest, ImageManagerGetItemsInfos)
 {
     ASSERT_TRUE(mDB.Init(mDatabaseConfig).IsNone());
 
-    auto item1 = CreateImageManagerItemInfo("service1", "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
-    auto item2 = CreateImageManagerItemInfo("service1", "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
-    auto item3 = CreateImageManagerItemInfo("service2", "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
+    auto item1 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "1.0.0", "sha256:abc123", ItemStateEnum::eInstalled);
+    auto item2 = CreateImageManagerItemInfo(
+        "service1", UpdateItemTypeEnum::eService, "2.0.0", "sha256:def456", ItemStateEnum::eInstalled);
+    auto item3 = CreateImageManagerItemInfo(
+        "service2", UpdateItemTypeEnum::eService, "1.0.0", "sha256:ghi789", ItemStateEnum::ePending);
 
     ASSERT_TRUE(mDB.AddItem(item1).IsNone());
     ASSERT_TRUE(mDB.AddItem(item2).IsNone());

src/cm/smcontroller/smhandler.cpp

@@ -38,9 +38,11 @@ SMHandler::SMHandler(grpc::ServerContext*
 
 void SMHandler::Start()
 {
+    std::lock_guard lock {mMutex};
+
     LOG_DBG() << "Start SM handler";
 
-    mStopProcessing.store(false);
+    mStopProcessing = false;
 
     mSyncMessageSender.Init(mStream, cResponseTime);
     mSyncMessageSender.RegisterResponseHandler(
@@ -54,25 +56,36 @@ void SMHandler::Start()
             dst.mutable_average_monitoring()->CopyFrom(src.average_monitoring());
         });
 
-    mProcessThread = std::thread([this]() { ProcessMessages(); });
+    mReadThread    = std::thread([this]() { ReadMessages(); });
+    mMessageThread = std::thread([this]() { ProcessMessages(); });
 }
 
 void SMHandler::Wait()
 {
-    if (mProcessThread.joinable()) {
-        mProcessThread.join();
+    LOG_DBG() << "Wait SM handler";
+
+    if (mReadThread.joinable()) {
+        mReadThread.join();
+    }
+
+    if (mMessageThread.joinable()) {
+        mMessageThread.join();
     }
 }
 
 void SMHandler::Stop()
 {
+    std::lock_guard lock {mMutex};
+
     LOG_DBG() << "Stop SM handler";
 
-    mStopProcessing.store(true);
+    mStopProcessing = true;
 
     if (mContext) {
         mContext->TryCancel();
     }
+
+    mCondVar.notify_one();
 }
 
 String SMHandler::GetNodeID() const
@@ -213,13 +226,56 @@ Error SMHandler::UpdateNetworks(const Array<UpdateNetworkParameters>& networkPar
  * Private
  **********************************************************************************************************************/
 
-Error SMHandler::ProcessMessages()
+void SMHandler::ReadMessages()
 {
     try {
         servicemanager::v5::SMOutgoingMessages outgoingMsg;
 
-        while (!mStopProcessing.load() && mStream->Read(&outgoingMsg)) {
+        while (mStream->Read(&outgoingMsg)) {
+            if (auto err = mSyncMessageSender.ProcessResponse(outgoingMsg); err.HasValue()) {
+                if (!err->IsNone()) {
+                    LOG_ERR() << "Failed to process response" << Log::Field("nodeID", GetNodeID())
+                              << Log::Field(AOS_ERROR_WRAP(*err));
+                }
+            } else {
+                std::lock_guard lock {mMutex};
+
+                mMessageQueue.push(outgoingMsg);
+                mCondVar.notify_one();
+            }
+        }
+    } catch (const std::exception& e) {
+        LOG_ERR() << "Handle incoming messages failed" << Log::Field(AOS_ERROR_WRAP(common::utils::ToAosError(e)));
+    }
+
+    mConnStatusListener->OnNodeDisconnected(GetNodeID());
+
+    std::lock_guard lock {mMutex};
+
+    mStopProcessing = true;
+    mCondVar.notify_one();
+}
+
+void SMHandler::ProcessMessages()
+{
+    while (true) {
+        try {
+            std::unique_lock lock {mMutex};
+
+            mCondVar.wait(lock, [this]() { return mStopProcessing || !mMessageQueue.empty(); });
+
+            if (mStopProcessing) {
+                break;
+            }
+
             Error err;
+            auto  outgoingMsg = mMessageQueue.front();
+
+            mMessageQueue.pop();
+
+            // Process message without holding the lock to allow sending new messages in parallel
+
+            lock.unlock();
 
             if (outgoingMsg.has_sm_info()) {
                 err = ProcessSMInfo(outgoingMsg.sm_info());
@@ -233,23 +289,18 @@ Error SMHandler::ProcessMessages()
                 err = ProcessInstantMonitoring(outgoingMsg.instant_monitoring());
             } else if (outgoingMsg.has_alert()) {
                 err = ProcessAlert(outgoingMsg.alert());
-            } else if (auto processErr = mSyncMessageSender.ProcessResponse(outgoingMsg); processErr.HasValue()) {
-                err = processErr.GetValue();
             } else {
                 LOG_WRN() << "Unknown message type received";
             }
 
             if (!err.IsNone()) {
                 LOG_ERR() << "Failed to process message" << Log::Field("nodeID", GetNodeID()) << Log::Field(err);
             }
+        } catch (const std::exception& e) {
+            LOG_ERR() << "Process message failed" << Log::Field("nodeID", GetNodeID())
+                      << Log::Field(AOS_ERROR_WRAP(common::utils::ToAosError(e)));
         }
-    } catch (const std::exception& e) {
-        LOG_ERR() << "Handle incoming messages failed" << Log::Field(AOS_ERROR_WRAP(common::utils::ToAosError(e)));
     }
-
-    mConnStatusListener->OnNodeDisconnected(GetNodeID());
-
-    return ErrorEnum::eNone;
 }
 
 Error SMHandler::SendMessage(const servicemanager::v5::SMIncomingMessages& message)