[ICD] Server side subscription persistence and resumption

src/app/BUILD.gn

@@ -173,8 +173,11 @@ static_library("app") {
     "ReadHandler.cpp",
     "RequiredPrivilege.cpp",
     "RequiredPrivilege.h",
+    "SimpleSubscriptionResumptionStorage.cpp",
+    "SimpleSubscriptionResumptionStorage.h",
     "StatusResponse.cpp",
     "StatusResponse.h",
+    "SubscriptionResumptionStorage.h",
     "TimedHandler.cpp",
     "TimedHandler.h",
     "TimedRequest.cpp",

src/app/InteractionModelEngine.cpp

@@ -50,14 +50,16 @@ InteractionModelEngine * InteractionModelEngine::GetInstance()
 }
 
 CHIP_ERROR InteractionModelEngine::Init(Messaging::ExchangeManager * apExchangeMgr, FabricTable * apFabricTable,
-                                        CASESessionManager * apCASESessionMgr)
+                                        CASESessionManager * apCASESessionMgr,
+                                        SubscriptionResumptionStorage * subscriptionResumptionStorage)
 {
     VerifyOrReturnError(apFabricTable != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
     VerifyOrReturnError(apExchangeMgr != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
 
-    mpExchangeMgr    = apExchangeMgr;
-    mpFabricTable    = apFabricTable;
-    mpCASESessionMgr = apCASESessionMgr;
+    mpExchangeMgr                   = apExchangeMgr;
+    mpFabricTable                   = apFabricTable;
+    mpCASESessionMgr                = apCASESessionMgr;
+    mpSubscriptionResumptionStorage = subscriptionResumptionStorage;
 
     ReturnErrorOnFailure(mpFabricTable->AddFabricDelegate(this));
     ReturnErrorOnFailure(mpExchangeMgr->RegisterUnsolicitedMessageHandlerForProtocol(Protocols::InteractionModel::Id, this));
@@ -664,6 +666,8 @@ Status InteractionModelEngine::OnUnsolicitedReportData(Messaging::ExchangeContex
         return Status::Success;
     }
 
+    ChipLogDetail(InteractionModel, "Received report with invalid subscriptionId %" PRIu32, subscriptionId);
+
     return Status::InvalidSubscription;
 }
 
@@ -1578,5 +1582,40 @@ void InteractionModelEngine::OnFabricRemoved(const FabricTable & fabricTable, Fa
     // the fabric removal, though, so they will fail when they try to actually send their command response
     // and will close at that point.
 }
+
+CHIP_ERROR InteractionModelEngine::ResumeSubscriptions()
+{
+#if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+    ReturnErrorCodeIf(!mpSubscriptionResumptionStorage, CHIP_NO_ERROR);
+
+    SubscriptionResumptionStorage::SubscriptionInfo subscriptionInfo;
+    auto * iterator = mpSubscriptionResumptionStorage->IterateSubscriptions();
+    while (iterator->Next(subscriptionInfo))
+    {
+        auto requestedAttributePathCount = subscriptionInfo.mAttributePaths.AllocatedSize();
+        auto requestedEventPathCount     = subscriptionInfo.mEventPaths.AllocatedSize();
+        if (!EnsureResourceForSubscription(subscriptionInfo.mFabricIndex, requestedAttributePathCount, requestedEventPathCount))
+        {
+            ChipLogProgress(InteractionModel, "no resource for Subscription resumption");
+            iterator->Release();
+            return CHIP_ERROR_NO_MEMORY;
+        }
+
+        ReadHandler * handler = mReadHandlers.CreateObject(*this);
+        if (handler == nullptr)
+        {
+            ChipLogProgress(InteractionModel, "no resource for ReadHandler creation");
+            iterator->Release();
+            return CHIP_ERROR_NO_MEMORY;
+        }
+
+        handler->ResumeSubscription(*mpCASESessionMgr, subscriptionInfo);
+    }
+    iterator->Release();
+#endif // CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+
+    return CHIP_NO_ERROR;
+}
+
 } // namespace app
 } // namespace chip

src/app/InteractionModelEngine.h

@@ -114,7 +114,8 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
      *
      */
     CHIP_ERROR Init(Messaging::ExchangeManager * apExchangeMgr, FabricTable * apFabricTable,
-                    CASESessionManager * apCASESessionMgr = nullptr);
+                    CASESessionManager * apCASESessionMgr                         = nullptr,
+                    SubscriptionResumptionStorage * subscriptionResumptionStorage = nullptr);
 
     void Shutdown();
 
@@ -292,6 +293,10 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
     // virtual method from FabricTable::Delegate
     void OnFabricRemoved(const FabricTable & fabricTable, FabricIndex fabricIndex) override;
 
+    SubscriptionResumptionStorage * GetSubscriptionResumptionStorage() { return mpSubscriptionResumptionStorage; };
+
+    CHIP_ERROR ResumeSubscriptions();
+
 #if CONFIG_BUILD_FOR_HOST_UNIT_TEST
     //
     // Get direct access to the underlying read handler pool
@@ -596,6 +601,8 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
 
     CASESessionManager * mpCASESessionMgr = nullptr;
 
+    SubscriptionResumptionStorage * mpSubscriptionResumptionStorage = nullptr;
+
     // A magic number for tracking values between stack Shutdown()-s and Init()-s.
     // An ObjectHandle is valid iff. its magic equals to this one.
     uint32_t mMagic = 0;

src/app/ReadHandler.cpp

@@ -41,7 +41,8 @@ using Status = Protocols::InteractionModel::Status;
 ReadHandler::ReadHandler(ManagementCallback & apCallback, Messaging::ExchangeContext * apExchangeContext,
                          InteractionType aInteractionType) :
     mExchangeCtx(*this),
-    mManagementCallback(apCallback)
+    mManagementCallback(apCallback), mOnConnectedCallback(HandleDeviceConnected, this),
+    mOnConnectionFailureCallback(HandleDeviceConnectionFailure, this)
 {
     VerifyOrDie(apExchangeContext != nullptr);
 
@@ -61,6 +62,54 @@ ReadHandler::ReadHandler(ManagementCallback & apCallback, Messaging::ExchangeCon
     mSessionHandle.Grab(mExchangeCtx->GetSessionHandle());
 }
 
+#if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+ReadHandler::ReadHandler(ManagementCallback & apCallback) :
+    mExchangeCtx(*this), mManagementCallback(apCallback), mOnConnectedCallback(HandleDeviceConnected, this),
+    mOnConnectionFailureCallback(HandleDeviceConnectionFailure, this)
+{
+    mInteractionType = InteractionType::Subscribe;
+    mFlags.ClearAll();
+}
+
+void ReadHandler::ResumeSubscription(CASESessionManager & caseSessionManager,
+                                     SubscriptionResumptionStorage::SubscriptionInfo & subscriptionInfo)
+{
+    mSubscriptionId          = subscriptionInfo.mSubscriptionId;
+    mMinIntervalFloorSeconds = subscriptionInfo.mMinInterval;
+    mMaxInterval             = subscriptionInfo.mMaxInterval;
+    SetStateFlag(ReadHandlerFlags::FabricFiltered, subscriptionInfo.mFabricFiltered);
+
+    // Move dynamically allocated attributes and events from the SubscriptionInfo struct into
+    // the object pool managed by the IM engine
+    for (size_t i = 0; i < subscriptionInfo.mAttributePaths.AllocatedSize(); i++)
+    {
+        AttributePathParams attributePathParams = subscriptionInfo.mAttributePaths[i].GetParams();
+        CHIP_ERROR err =
+            InteractionModelEngine::GetInstance()->PushFrontAttributePathList(mpAttributePathList, attributePathParams);
+        if (err != CHIP_NO_ERROR)
+        {
+            Close();
+            return;
+        }
+    }
+    for (size_t i = 0; i < subscriptionInfo.mEventPaths.AllocatedSize(); i++)
+    {
+        EventPathParams eventPathParams = subscriptionInfo.mEventPaths[i].GetParams();
+        CHIP_ERROR err = InteractionModelEngine::GetInstance()->PushFrontEventPathParamsList(mpEventPathList, eventPathParams);
+        if (err != CHIP_NO_ERROR)
+        {
+            Close();
+            return;
+        }
+    }
+
+    // Ask IM engine to start CASE session with subscriber
+    ScopedNodeId peerNode = ScopedNodeId(subscriptionInfo.mNodeId, subscriptionInfo.mFabricIndex);
+    caseSessionManager.FindOrEstablishSession(peerNode, &mOnConnectedCallback, &mOnConnectionFailureCallback);
+}
+
+#endif // CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+
 ReadHandler::~ReadHandler()
 {
     auto * appCallback = mManagementCallback.GetAppCallback();
@@ -87,8 +136,18 @@ ReadHandler::~ReadHandler()
     InteractionModelEngine::GetInstance()->ReleaseDataVersionFilterList(mpDataVersionFilterList);
 }
 
-void ReadHandler::Close()
+void ReadHandler::Close(CloseOptions options)
 {
+#if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+    if (options == CloseOptions::kDropPersistedSubscription)
+    {
+        auto * subscriptionResumptionStorage = InteractionModelEngine::GetInstance()->GetSubscriptionResumptionStorage();
+        if (subscriptionResumptionStorage)
+        {
+            subscriptionResumptionStorage->Delete(GetInitiatorNodeId(), GetAccessingFabricIndex(), mSubscriptionId);
+        }
+    }
+#endif // CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
     MoveToState(HandlerState::AwaitingDestruction);
     mManagementCallback.OnDone(*this);
 }
@@ -306,7 +365,12 @@ void ReadHandler::OnResponseTimeout(Messaging::ExchangeContext * apExchangeConte
 {
     ChipLogError(DataManagement, "Time out! failed to receive status response from Exchange: " ChipLogFormatExchange,
                  ChipLogValueExchange(apExchangeContext));
+#if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+    // TODO: Have a retry mechanism tied to wake interval for IC devices
+    Close(CloseOptions::kKeepPersistedSubscription);
+#else
     Close();
+#endif
 }
 
 CHIP_ERROR ReadHandler::ProcessReadRequest(System::PacketBufferHandle && aPayload)
@@ -652,9 +716,59 @@ CHIP_ERROR ReadHandler::ProcessSubscribeRequest(System::PacketBufferHandle && aP
 
     mExchangeCtx->WillSendMessage();
 
+#if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+    PersistSubscription();
+#endif // CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
+
     return CHIP_NO_ERROR;
 }
 
+void ReadHandler::PersistSubscription()
+{
+    auto * subscriptionResumptionStorage = InteractionModelEngine::GetInstance()->GetSubscriptionResumptionStorage();
+    VerifyOrReturn(subscriptionResumptionStorage != nullptr);
+
+    SubscriptionResumptionStorage::SubscriptionInfo subscriptionInfo = { .mNodeId         = GetInitiatorNodeId(),
+                                                                         .mFabricIndex    = GetAccessingFabricIndex(),
+                                                                         .mSubscriptionId = mSubscriptionId,
+                                                                         .mMinInterval    = mMinIntervalFloorSeconds,
+                                                                         .mMaxInterval    = mMaxInterval,
+                                                                         .mFabricFiltered = IsFabricFiltered() };
+    ObjectList<AttributePathParams> * attributePath                  = mpAttributePathList;
+    size_t attributePathCount                                        = 0;
+    while (attributePath)
+    {
+        attributePathCount++;
+        attributePath = attributePath->mpNext;
+    }
+    attributePath = mpAttributePathList;
+    subscriptionInfo.mAttributePaths.Calloc(attributePathCount);
+    VerifyOrReturn(subscriptionInfo.mAttributePaths.Get() != nullptr);
+    for (size_t i = 0; i < attributePathCount; i++)
+    {
+        subscriptionInfo.mAttributePaths[i].SetValues(attributePath->mValue);
+        attributePath = attributePath->mpNext;
+    }
+
+    ObjectList<EventPathParams> * eventPath = mpEventPathList;
+    size_t eventPathCount                   = 0;
+    while (eventPath)
+    {
+        eventPathCount++;
+        eventPath = eventPath->mpNext;
+    }
+    eventPath = mpEventPathList;
+    subscriptionInfo.mEventPaths.Calloc(eventPathCount);
+    VerifyOrReturn(subscriptionInfo.mEventPaths.Get() != nullptr);
+    for (size_t i = 0; i < eventPathCount; i++)
+    {
+        subscriptionInfo.mEventPaths[i].SetValues(eventPath->mValue);
+        eventPath = eventPath->mpNext;
+    }
+
+    subscriptionResumptionStorage->Save(subscriptionInfo);
+}
+
 void ReadHandler::OnUnblockHoldReportCallback(System::Layer * apSystemLayer, void * apAppState)
 {
     VerifyOrReturn(apAppState != nullptr);
@@ -767,5 +881,33 @@ void ReadHandler::ClearStateFlag(ReadHandlerFlags aFlag)
     SetStateFlag(aFlag, false);
 }
 
+void ReadHandler::HandleDeviceConnected(void * context, Messaging::ExchangeManager & exchangeMgr,
+                                        const SessionHandle & sessionHandle)
+{
+    ReadHandler * const _this = static_cast<ReadHandler *>(context);
+
+    _this->mSessionHandle.Grab(sessionHandle);
+
+    _this->MoveToState(HandlerState::GeneratingReports);
+
+    ObjectList<AttributePathParams> * attributePath = _this->mpAttributePathList;
+    while (attributePath)
+    {
+        InteractionModelEngine::GetInstance()->GetReportingEngine().SetDirty(attributePath->mValue);
+        attributePath = attributePath->mpNext;
+    }
+}
+
+void ReadHandler::HandleDeviceConnectionFailure(void * context, const ScopedNodeId & peerId, CHIP_ERROR err)
+{
+    ReadHandler * const _this = static_cast<ReadHandler *>(context);
+    VerifyOrDie(_this != nullptr);
+
+    // TODO: Have a retry mechanism tied to wake interval for IC devices
+    ChipLogError(DataManagement, "Failed to establish CASE for subscription-resumption with error '%" CHIP_ERROR_FORMAT "'",
+                 err.Format());
+    _this->Close();
+}
+
 } // namespace app
 } // namespace chip