Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ICD]Trigger resubscription when receiving check-in and subscription has not yet become abnormal #37831

Open
wants to merge 10 commits into
base: master
Choose a base branch
from
Open

[ICD]Trigger resubscription when receiving check-in and subscription has not yet become abnormal #37831

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
9f7bed7
Trigger resubscription when receiving check-in and subscription has n…
yunhanw-google Mar 1, 2025
2a2df3e
fix OnPeerTypeChange
yunhanw-google Mar 1, 2025
e49f9a6
address comments
yunhanw-google Mar 7, 2025
8dce4be
add monitor subject check for subscription activation
yunhanw-google Mar 8, 2025
0860870
fix compilation
yunhanw-google Mar 8, 2025
9cda649
introduce mCatsMatchCheckIn to indicate whether cats for icd registra…
yunhanw-google Mar 10, 2025
287a093
address comments and add tests
yunhanw-google Mar 10, 2025
2b29328
Restyled by whitespace
restyled-commits Mar 10, 2025
6a97e2c
Restyled by clang-format
restyled-commits Mar 10, 2025
1fca04a
address comments
yunhanw-google Mar 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions src/app/InteractionModelEngine.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,8 +236,7 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
/**
* Activate the idle subscriptions.
*
* When subscribing to ICD and liveness timeout reached, the read client will move to `InactiveICDSubscription` state and
* resubscription can be triggered via OnActiveModeNotification().
* See ReadClient::OnActiveModeNotification
*/
void OnActiveModeNotification(ScopedNodeId aPeer);

Expand Down
28 changes: 22 additions & 6 deletions src/app/ReadClient.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -485,6 +485,15 @@ CHIP_ERROR ReadClient::GenerateDataVersionFilterList(DataVersionFilterIBs::Build
void ReadClient::OnActiveModeNotification()
{
VerifyOrDie(mpImEngine->InActiveReadClientList(this));

// Note: this API only works when issuing subscription via SendAutoResubscribeRequest. When SendAutoResubscribeRequest is
// called, either mEventPathParamsListSize or mAttributePathParamsListSize is not 0.
VerifyOrReturn(mReadPrepareParams.mEventPathParamsListSize != 0 || mReadPrepareParams.mAttributePathParamsListSize != 0);

// If mCatsMatchCheckIn is true, it means cats used in icd registration matches with the one in current subscription,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Matches which side? Should be "our side of the current subscription" or something, right?

But do we really want to push this off onto the API client? Or just examine the fabric table?

And if we do push this off onto the API client, as this PR does, then should this boolean just be mResetSubscriptionOnCheckin?

// OnActiveModeNotification continues to revive the subscription as needed, otherwise, do nothing.
VerifyOrReturn(mReadPrepareParams.mCatsMatchCheckIn);

// When we reach here, the subscription definitely exceeded the liveness timeout. Just continue the unfinished resubscription
// logic in `OnLivenessTimeoutCallback`.
if (IsInactiveICDSubscription())
Expand All @@ -493,6 +502,19 @@ void ReadClient::OnActiveModeNotification()
return;
}

// If the server sends out check-in message, and there is a tracked active subscription in client side at the same time, it
// means current client does not realize this tracked subscription has gone, and we should forcibly timeout current
// subscription, and schedule a new one.
if (!mIsResubscriptionScheduled)
{
// Closing will ultimately trigger ScheduleResubscription with the aReestablishCASE argument set to true, effectively
// rendering the session defunct.
Close(CHIP_ERROR_TIMEOUT);
return;
}

// If the server sends a check-in message and a subscription is already scheduled, it indicates a client-side subscription
// timeout or failure. Cancel the scheduled subscription and initiate a new one immediately.
TriggerResubscribeIfScheduled("check-in message");
}

Expand All @@ -503,12 +525,6 @@ void ReadClient::OnPeerTypeChange(PeerType aType)
mIsPeerLIT = (aType == PeerType::kLITICD);

ChipLogProgress(DataManagement, "Peer is now %s LIT ICD.", mIsPeerLIT ? "a" : "not a");

// If the peer is no longer LIT, try to wake up the subscription and do resubscribe when necessary.
if (!mIsPeerLIT)
{
OnActiveModeNotification();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So I am trying to understand why it's OK to remove this. The reason this seems to be here is that it's envisioning us having two subscriptions, one of which (call it A) is watching for peer type change and one of which (call it B) is IsInactiveICDSubscription(). When the transition to SIT is observed, B will no longer get any OnActiveModeNotifications, so how will it get out of the "inactive ICD subscription" state?

Of course if A does not exist and we just have B, and it's in the "inactive ICD subscription" state and then the server changes to A SIT, I think we just get stuck; that seems to be a hole in the whole logic of this stuff, including the spec?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right, the current change does not work on the below scenario

So I am trying to understand why it's OK to remove this. The reason this seems to be here is that it's envisioning us having two subscriptions, one of which (call it A) is watching for peer type change and one of which (call it B) is IsInactiveICDSubscription(). When the transition to SIT is observed, B will no longer get any OnActiveModeNotifications, so how will it get out of the "inactive ICD subscription" state?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated the code.

  1. Update PeerTypeChange and call TriggerResubscriptionForLivenessTimeout when peer type is changed to LIT and state is inactive icd subscription, instead of OnActiveModeNotification to help situation 1 in your question.
  2. If icd transition from lit to sit, it would send check-in message, which would bring up the in active subscription.

}
}

CHIP_ERROR ReadClient::OnMessageReceived(Messaging::ExchangeContext * apExchangeContext, const PayloadHeader & aPayloadHeader,
Expand Down
12 changes: 9 additions & 3 deletions src/app/ReadClient.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,11 +351,17 @@ class ReadClient : public Messaging::ExchangeDelegate
* Re-activate an inactive subscription.
*
* When subscribing to LIT-ICD and liveness timeout reached and OnResubscriptionNeeded returns
* CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT, the read client will move to the InactiveICDSubscription state and
* resubscription can be triggered via OnActiveModeNotification().
* CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT, the read client will move to the InactiveICDSubscription state and
* resubscription can be triggered via OnActiveModeNotification().
*
* If the subscription is not in the `InactiveICDSubscription` state, this function will do nothing. So it is always safe to
* call this function when a check-in message is received.
* call this function when a check-in message is received.
*
* If the server sends out check-in message, and there is a active tracked active subscription in client side at the same time,
* it means current client does not realize this tracked subscription has gone, and we should forcibly timeout current
* subscription, and schedule a new one.
Comment on lines +360 to +362
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment does not match what the API actually does, really.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

rewrite the comments, thanks

*
* This API only works when issuing subscription via SendAutoResubscribeRequest.
*/
void OnActiveModeNotification();

Expand Down
5 changes: 5 additions & 0 deletions src/app/ReadPrepareParams.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,9 @@ struct ReadPrepareParams
bool mIsFabricFiltered = true;
bool mIsPeerLIT = false;

// see ReadClient::OnActiveModeNotification
bool mCatsMatchCheckIn = true;

ReadPrepareParams() {}
ReadPrepareParams(const SessionHandle & sessionHandle) { mSessionHolder.Grab(sessionHandle); }
ReadPrepareParams(ReadPrepareParams && other) : mSessionHolder(other.mSessionHolder)
Expand All @@ -66,6 +69,7 @@ struct ReadPrepareParams
mTimeout = other.mTimeout;
mIsFabricFiltered = other.mIsFabricFiltered;
mIsPeerLIT = other.mIsPeerLIT;
mCatsMatchCheckIn = other.mCatsMatchCheckIn;
other.mpEventPathParamsList = nullptr;
other.mEventPathParamsListSize = 0;
other.mpAttributePathParamsList = nullptr;
Expand All @@ -91,6 +95,7 @@ struct ReadPrepareParams
mTimeout = other.mTimeout;
mIsFabricFiltered = other.mIsFabricFiltered;
mIsPeerLIT = other.mIsPeerLIT;
mCatsMatchCheckIn = other.mCatsMatchCheckIn;
other.mpEventPathParamsList = nullptr;
other.mEventPathParamsListSize = 0;
other.mpAttributePathParamsList = nullptr;
Expand Down
133 changes: 133 additions & 0 deletions src/controller/tests/data_model/TestRead.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2354,6 +2354,139 @@ TEST_F(TestRead, TestSubscribe_OnActiveModeNotification)
EXPECT_EQ(GetExchangeManager().GetNumActiveExchanges(), 0u);
}

/**
* When all tracked subscriptions go away in server, check-in message is received and OnActiveModeNotification is called in client
* side, the tracked subscription would be torn down and a new one would be rescheduled in client side.
*/
TEST_F(TestRead, TestSubscribe_SubGoAwayInserverOnActiveModeNotification)
{
auto sessionHandle = GetSessionBobToAlice();

SetMRPMode(MessagingContext::MRPMode::kResponsive);

{
TestResubscriptionCallback callback;
ReadClient readClient(InteractionModelEngine::GetInstance(), &GetExchangeManager(), callback,
ReadClient::InteractionType::Subscribe);

callback.mScheduleLITResubscribeImmediately = false;
callback.SetReadClient(&readClient);

ReadPrepareParams readPrepareParams(GetSessionBobToAlice());

// Read full wildcard paths, repeat twice to ensure chunking.
AttributePathParams attributePathParams[1];
readPrepareParams.mpAttributePathParamsList = attributePathParams;
readPrepareParams.mAttributePathParamsListSize = MATTER_ARRAY_SIZE(attributePathParams);
attributePathParams[0].mEndpointId = kTestEndpointId;
attributePathParams[0].mClusterId = Clusters::UnitTesting::Id;
attributePathParams[0].mAttributeId = Clusters::UnitTesting::Attributes::Boolean::Id;

constexpr uint16_t maxIntervalCeilingSeconds = 1;

readPrepareParams.mMaxIntervalCeilingSeconds = maxIntervalCeilingSeconds;
readPrepareParams.mIsPeerLIT = true;

auto err = readClient.SendAutoResubscribeRequest(std::move(readPrepareParams));
EXPECT_EQ(err, CHIP_NO_ERROR);

//
// Drive servicing IO till we have established a subscription.
//
GetIOContext().DriveIOUntil(System::Clock::Milliseconds32(2000),
[&]() { return callback.mOnSubscriptionEstablishedCount >= 1; });
EXPECT_EQ(callback.mOnSubscriptionEstablishedCount, 1);
EXPECT_EQ(callback.mOnError, 0);
EXPECT_EQ(callback.mOnResubscriptionsAttempted, 0);

GetLoopback().mNumMessagesToDrop = 0;
callback.ClearCounters();
InteractionModelEngine::GetInstance()->OnActiveModeNotification(
ScopedNodeId(readClient.GetPeerNodeId(), readClient.GetFabricIndex()));
EXPECT_EQ(callback.mOnResubscriptionsAttempted, 1);
EXPECT_EQ(callback.mLastError, CHIP_ERROR_TIMEOUT);

//
// Drive servicing IO till we have established a subscription.
//
GetIOContext().DriveIOUntil(System::Clock::Milliseconds32(2000),
[&]() { return callback.mOnSubscriptionEstablishedCount == 1; });
EXPECT_EQ(callback.mOnSubscriptionEstablishedCount, 1);

//
// With re-sub enabled, we shouldn't have encountered any errors
//
EXPECT_EQ(callback.mOnError, 0);
EXPECT_EQ(callback.mOnDone, 0);
}

SetMRPMode(MessagingContext::MRPMode::kDefault);

InteractionModelEngine::GetInstance()->ShutdownActiveReads();
EXPECT_EQ(GetExchangeManager().GetNumActiveExchanges(), 0u);
}

/**
* When all tracked subscriptions go away in server, check-in message is received and OnActiveModeNotification is called in client
* side, the untracked subscription would be kept.
*/
TEST_F(TestRead, TestSubscribe_MismatchedSubGoAwayInserverOnActiveModeNotification)
{
auto sessionHandle = GetSessionBobToAlice();

SetMRPMode(MessagingContext::MRPMode::kResponsive);

{
TestResubscriptionCallback callback;
ReadClient readClient(InteractionModelEngine::GetInstance(), &GetExchangeManager(), callback,
ReadClient::InteractionType::Subscribe);

callback.mScheduleLITResubscribeImmediately = false;
callback.SetReadClient(&readClient);

ReadPrepareParams readPrepareParams(GetSessionBobToAlice());

// Read full wildcard paths, repeat twice to ensure chunking.
AttributePathParams attributePathParams[1];
readPrepareParams.mpAttributePathParamsList = attributePathParams;
readPrepareParams.mAttributePathParamsListSize = MATTER_ARRAY_SIZE(attributePathParams);
attributePathParams[0].mEndpointId = kTestEndpointId;
attributePathParams[0].mClusterId = Clusters::UnitTesting::Id;
attributePathParams[0].mAttributeId = Clusters::UnitTesting::Attributes::Boolean::Id;
readPrepareParams.mCatsMatchCheckIn = false;
constexpr uint16_t maxIntervalCeilingSeconds = 1;

readPrepareParams.mMaxIntervalCeilingSeconds = maxIntervalCeilingSeconds;
readPrepareParams.mIsPeerLIT = true;

auto err = readClient.SendAutoResubscribeRequest(std::move(readPrepareParams));
EXPECT_EQ(err, CHIP_NO_ERROR);

//
// Drive servicing IO till we have established a subscription.
//
GetIOContext().DriveIOUntil(System::Clock::Milliseconds32(2000),
[&]() { return callback.mOnSubscriptionEstablishedCount >= 1; });
EXPECT_EQ(callback.mOnSubscriptionEstablishedCount, 1);
EXPECT_EQ(callback.mOnError, 0);
EXPECT_EQ(callback.mOnResubscriptionsAttempted, 0);

GetLoopback().mNumMessagesToDrop = 0;
callback.ClearCounters();
InteractionModelEngine::GetInstance()->OnActiveModeNotification(
ScopedNodeId(readClient.GetPeerNodeId(), readClient.GetFabricIndex()));
EXPECT_EQ(callback.mOnResubscriptionsAttempted, 0);
EXPECT_EQ(callback.mLastError, CHIP_NO_ERROR);
EXPECT_EQ(callback.mOnError, 0);
EXPECT_EQ(callback.mOnDone, 0);
}

SetMRPMode(MessagingContext::MRPMode::kDefault);

InteractionModelEngine::GetInstance()->ShutdownActiveReads();
EXPECT_EQ(GetExchangeManager().GetNumActiveExchanges(), 0u);
}

TEST_F(TestRead, TestSubscribeFailed_OnActiveModeNotification)
{
auto sessionHandle = GetSessionBobToAlice();
Expand Down
Loading