Better fix for crashes around MTRBaseSubscriptionCallback. (project-chip#23076)

project-chip#22978 accidentally
reintroduced the crash that
project-chip#22324 had fixed.  To avoid
more issues along these lines:

1) Add unit tests that reproduce the crashes described in
   project-chip#22320 (with the
   changes from project-chip#22978) and
   project-chip#22935 (without those
   changes).
2) Change MTRBaseSubscriptionCallback to always invoke its callbacks
   synchronously, on the Matter queue, so that we can clean up the
   MTRClusterStateCacheContainer's pointer to the ClusterStateCache before it
   gets deleted on the Matter queue.
3) Move the queueing of callbacks to the client queue into the consumers of
   MTRBaseSubscriptionCallback, so they can do whatever sync work they need
   (like the above cleanup) before going async.
4) Update documentation.

Author: bzbarsky-apple

src/darwin/Framework/CHIP/MTRBaseDevice.mm

@@ -251,11 +251,11 @@ - (void)invalidateCASESession
 
 class SubscriptionCallback final : public MTRBaseSubscriptionCallback {
 public:
-    SubscriptionCallback(dispatch_queue_t queue, DataReportCallback attributeReportCallback, DataReportCallback eventReportCallback,
+    SubscriptionCallback(DataReportCallback attributeReportCallback, DataReportCallback eventReportCallback,
         ErrorCallback errorCallback, MTRDeviceResubscriptionScheduledHandler _Nullable resubscriptionScheduledHandler,
         MTRSubscriptionEstablishedHandler _Nullable subscriptionEstablishedHandler, OnDoneHandler _Nullable onDoneHandler)
-        : MTRBaseSubscriptionCallback(queue, attributeReportCallback, eventReportCallback, errorCallback,
-            resubscriptionScheduledHandler, subscriptionEstablishedHandler, onDoneHandler)
+        : MTRBaseSubscriptionCallback(attributeReportCallback, eventReportCallback, errorCallback, resubscriptionScheduledHandler,
+            subscriptionEstablishedHandler, onDoneHandler)
     {
     }
 
@@ -286,80 +286,120 @@ - (void)subscribeWithQueue:(dispatch_queue_t)queue
     // Copy params before going async.
     params = [params copy];
 
-    [self.deviceController
-        getSessionForNode:self.nodeID
-               completion:^(ExchangeManager * _Nullable exchangeManager, const Optional<SessionHandle> & session,
-                   NSError * _Nullable error) {
-                   if (error != nil) {
-                       dispatch_async(queue, ^{
-                           errorHandler(error);
-                       });
-                       return;
-                   }
-
-                   // Wildcard endpoint, cluster, attribute, event.
-                   auto attributePath = std::make_unique<AttributePathParams>();
-                   auto eventPath = std::make_unique<EventPathParams>();
-                   ReadPrepareParams readParams(session.Value());
-                   readParams.mMinIntervalFloorSeconds = [params.minInterval unsignedShortValue];
-                   readParams.mMaxIntervalCeilingSeconds = [params.maxInterval unsignedShortValue];
-                   readParams.mpAttributePathParamsList = attributePath.get();
-                   readParams.mAttributePathParamsListSize = 1;
-                   readParams.mpEventPathParamsList = eventPath.get();
-                   readParams.mEventPathParamsListSize = 1;
-                   readParams.mIsFabricFiltered = params.fabricFiltered;
-                   readParams.mKeepSubscriptions = params.keepPreviousSubscriptions;
-
-                   std::unique_ptr<SubscriptionCallback> callback;
-                   std::unique_ptr<ReadClient> readClient;
-                   std::unique_ptr<ClusterStateCache> clusterStateCache;
-                   if (clusterStateCacheContainer) {
-                       __weak MTRClusterStateCacheContainer * weakPtr = clusterStateCacheContainer;
-                       callback = std::make_unique<SubscriptionCallback>(queue, attributeReportHandler, eventReportHandler,
-                           errorHandler, resubscriptionScheduled, subscriptionEstablished, ^{
-                               MTRClusterStateCacheContainer * container = weakPtr;
-                               if (container) {
-                                   container.cppClusterStateCache = nullptr;
-                               }
-                           });
-                       clusterStateCache = std::make_unique<ClusterStateCache>(*callback.get());
-                       readClient = std::make_unique<ReadClient>(InteractionModelEngine::GetInstance(), exchangeManager,
-                           clusterStateCache->GetBufferedCallback(), ReadClient::InteractionType::Subscribe);
-                   } else {
-                       callback = std::make_unique<SubscriptionCallback>(queue, attributeReportHandler, eventReportHandler,
-                           errorHandler, resubscriptionScheduled, subscriptionEstablished, nil);
-                       readClient = std::make_unique<ReadClient>(InteractionModelEngine::GetInstance(), exchangeManager,
-                           callback->GetBufferedCallback(), ReadClient::InteractionType::Subscribe);
-                   }
-
-                   CHIP_ERROR err;
-                   if (!params.autoResubscribe) {
-                       err = readClient->SendRequest(readParams);
-                   } else {
-                       // SendAutoResubscribeRequest cleans up the params, even on failure.
-                       attributePath.release();
-                       eventPath.release();
-                       err = readClient->SendAutoResubscribeRequest(std::move(readParams));
-                   }
-
-                   if (err != CHIP_NO_ERROR) {
-                       dispatch_async(queue, ^{
-                           errorHandler([MTRError errorForCHIPErrorCode:err]);
-                       });
-
-                       return;
-                   }
-
-                   if (clusterStateCacheContainer) {
-                       clusterStateCacheContainer.cppClusterStateCache = clusterStateCache.get();
-                       // ClusterStateCache will be deleted when OnDone is called or an error is encountered as well.
-                       callback->AdoptClusterStateCache(std::move(clusterStateCache));
-                   }
-                   // Callback and ReadClient will be deleted when OnDone is called or an error is
-                   // encountered.
-                   callback->AdoptReadClient(std::move(readClient));
-                   callback.release();
-               }];
+    [self.deviceController getSessionForNode:self.nodeID
+                                  completion:^(ExchangeManager * _Nullable exchangeManager, const Optional<SessionHandle> & session,
+                                      NSError * _Nullable error) {
+                                      if (error != nil) {
+                                          dispatch_async(queue, ^{
+                                              errorHandler(error);
+                                          });
+                                          return;
+                                      }
+
+                                      // Wildcard endpoint, cluster, attribute, event.
+                                      auto attributePath = std::make_unique<AttributePathParams>();
+                                      auto eventPath = std::make_unique<EventPathParams>();
+                                      ReadPrepareParams readParams(session.Value());
+                                      readParams.mMinIntervalFloorSeconds = [params.minInterval unsignedShortValue];
+                                      readParams.mMaxIntervalCeilingSeconds = [params.maxInterval unsignedShortValue];
+                                      readParams.mpAttributePathParamsList = attributePath.get();
+                                      readParams.mAttributePathParamsListSize = 1;
+                                      readParams.mpEventPathParamsList = eventPath.get();
+                                      readParams.mEventPathParamsListSize = 1;
+                                      readParams.mIsFabricFiltered = params.fabricFiltered;
+                                      readParams.mKeepSubscriptions = params.keepPreviousSubscriptions;
+
+                                      std::unique_ptr<ClusterStateCache> clusterStateCache;
+                                      ReadClient::Callback * callbackForReadClient = nullptr;
+                                      OnDoneHandler onDoneHandler = nil;
+
+                                      if (clusterStateCacheContainer) {
+                                          __weak MTRClusterStateCacheContainer * weakPtr = clusterStateCacheContainer;
+                                          onDoneHandler = ^{
+                                              // This, like all manipulation of cppClusterStateCache, needs to run on the Matter
+                                              // queue.
+                                              MTRClusterStateCacheContainer * container = weakPtr;
+                                              if (container) {
+                                                  container.cppClusterStateCache = nullptr;
+                                              }
+                                          };
+                                      }
+
+                                      auto callback = std::make_unique<SubscriptionCallback>(
+                                          ^(NSArray * value) {
+                                              dispatch_async(queue, ^{
+                                                  if (attributeReportHandler != nil) {
+                                                      attributeReportHandler(value);
+                                                  }
+                                              });
+                                          },
+                                          ^(NSArray * value) {
+                                              dispatch_async(queue, ^{
+                                                  if (eventReportHandler != nil) {
+                                                      eventReportHandler(value);
+                                                  }
+                                              });
+                                          },
+                                          ^(NSError * error) {
+                                              dispatch_async(queue, ^{
+                                                  errorHandler(error);
+                                              });
+                                          },
+                                          ^(NSError * error, NSNumber * resubscriptionDelay) {
+                                              dispatch_async(queue, ^{
+                                                  if (resubscriptionScheduled != nil) {
+                                                      resubscriptionScheduled(error, resubscriptionDelay);
+                                                  }
+                                              });
+                                          },
+                                          ^(void) {
+                                              dispatch_async(queue, ^{
+                                                  if (subscriptionEstablished != nil) {
+                                                      subscriptionEstablished();
+                                                  }
+                                              });
+                                          },
+                                          onDoneHandler);
+
+                                      if (clusterStateCacheContainer) {
+                                          clusterStateCache = std::make_unique<ClusterStateCache>(*callback.get());
+                                          callbackForReadClient = &clusterStateCache->GetBufferedCallback();
+                                      } else {
+                                          callbackForReadClient = &callback->GetBufferedCallback();
+                                      }
+
+                                      auto readClient = std::make_unique<ReadClient>(InteractionModelEngine::GetInstance(),
+                                          exchangeManager, *callbackForReadClient, ReadClient::InteractionType::Subscribe);
+
+                                      CHIP_ERROR err;
+                                      if (!params.autoResubscribe) {
+                                          err = readClient->SendRequest(readParams);
+                                      } else {
+                                          // SendAutoResubscribeRequest cleans up the params, even on failure.
+                                          attributePath.release();
+                                          eventPath.release();
+                                          err = readClient->SendAutoResubscribeRequest(std::move(readParams));
+                                      }
+
+                                      if (err != CHIP_NO_ERROR) {
+                                          dispatch_async(queue, ^{
+                                              errorHandler([MTRError errorForCHIPErrorCode:err]);
+                                          });
+
+                                          return;
+                                      }
+
+                                      if (clusterStateCacheContainer) {
+                                          clusterStateCacheContainer.cppClusterStateCache = clusterStateCache.get();
+                                          // ClusterStateCache will be deleted when OnDone is called or an error is encountered as
+                                          // well.
+                                          callback->AdoptClusterStateCache(std::move(clusterStateCache));
+                                      }
+                                      // Callback and ReadClient will be deleted when OnDone is called or an error is
+                                      // encountered.
+                                      callback->AdoptReadClient(std::move(readClient));
+                                      callback.release();
+                                  }];
 }
 
 // Convert TLV data into data-value dictionary as described in MTRDeviceResponseHandler

src/darwin/Framework/CHIP/MTRBaseSubscriptionCallback.h

@@ -35,7 +35,16 @@
 /**
  * This file defines a base class for subscription callbacks used by
  * MTRBaseDevice and MTRDevice.  This base class handles everything except the
- * actual conversion from the incoming data to the desired data.
+ * actual conversion from the incoming data to the desired data and the dispatch
+ * of callbacks to the relevant client queues.  Its callbacks are called on the
+ * Matter queue.  This allows MTRDevice and MTRBaseDevice to do any necessary
+ * sync cleanup work before dispatching to the client callbacks on the client
+ * queue.
+ *
+ * After onDoneHandler is invoked, this object will at some point delete itself
+ * and destroy anything it owns (such as the ReadClient or the
+ * ClusterStateCache).  Consumers should drop references to all the relevant
+ * objects in that handler.  This deletion will happen on the Matter queue.
  *
  * The desired data is assumed to be NSObjects that can be stored in NSArray.
  */
@@ -49,12 +58,10 @@ typedef void (^OnDoneHandler)(void);
 
 class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callback {
 public:
-    MTRBaseSubscriptionCallback(dispatch_queue_t queue, DataReportCallback attributeReportCallback,
-        DataReportCallback eventReportCallback, ErrorCallback errorCallback,
-        MTRDeviceResubscriptionScheduledHandler _Nullable resubscriptionCallback,
+    MTRBaseSubscriptionCallback(DataReportCallback attributeReportCallback, DataReportCallback eventReportCallback,
+        ErrorCallback errorCallback, MTRDeviceResubscriptionScheduledHandler _Nullable resubscriptionCallback,
         SubscriptionEstablishedHandler _Nullable subscriptionEstablishedHandler, OnDoneHandler _Nullable onDoneHandler)
-        : mQueue(queue)
-        , mAttributeReportCallback(attributeReportCallback)
+        : mAttributeReportCallback(attributeReportCallback)
         , mEventReportCallback(eventReportCallback)
         , mErrorCallback(errorCallback)
         , mResubscriptionCallback(resubscriptionCallback)
@@ -117,10 +124,9 @@ class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callbac
     NSMutableArray * _Nullable mEventReports = nil;
 
 private:
-    dispatch_queue_t mQueue;
     DataReportCallback _Nullable mAttributeReportCallback = nil;
     DataReportCallback _Nullable mEventReportCallback = nil;
-    // We set mErrorCallback to nil when queueing error reports, so we
+    // We set mErrorCallback to nil before calling the error callback, so we
     // make sure to only report one error.
     ErrorCallback _Nullable mErrorCallback = nil;
     MTRDeviceResubscriptionScheduledHandler _Nullable mResubscriptionCallback = nil;
@@ -138,9 +144,11 @@ class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callbac
     // To handle this, enforce the following rules:
     //
     // 1) We guarantee that mErrorCallback is only invoked with an error once.
-    // 2) We ensure that we delete ourselves and the passed in ReadClient only from OnDone or a queued-up
-    //    error callback, but not both, by tracking whether we have a queued-up
-    //    deletion.
+    // 2) We guarantee that mOnDoneHandler is only invoked once, and always
+    //    invoked before we delete ourselves.
+    // 3) We ensure that we delete ourselves and the passed in ReadClient only
+    //    from OnDone or from an error callback but not both, by tracking whether
+    //    we have a queued-up deletion.
     std::unique_ptr<chip::app::ReadClient> mReadClient;
     std::unique_ptr<chip::app::ClusterStateCache> mClusterStateCache;
     bool mHaveQueuedDeletion = false;

src/darwin/Framework/CHIP/MTRBaseSubscriptionCallback.mm

@@ -33,21 +33,17 @@
 {
     __block NSArray * attributeReports = mAttributeReports;
     mAttributeReports = nil;
-    __block auto attributeCallback = mAttributeReportCallback;
+    auto attributeCallback = mAttributeReportCallback;
 
     __block NSArray * eventReports = mEventReports;
     mEventReports = nil;
-    __block auto eventCallback = mEventReportCallback;
+    auto eventCallback = mEventReportCallback;
 
     if (attributeCallback != nil && attributeReports.count) {
-        dispatch_async(mQueue, ^{
-            attributeCallback(attributeReports);
-        });
+        attributeCallback(attributeReports);
     }
     if (eventCallback != nil && eventReports.count) {
-        dispatch_async(mQueue, ^{
-            eventCallback(eventReports);
-        });
+        eventCallback(eventReports);
     }
 }
 
@@ -96,7 +92,8 @@
 void MTRBaseSubscriptionCallback::OnSubscriptionEstablished(SubscriptionId aSubscriptionId)
 {
     if (mSubscriptionEstablishedHandler) {
-        dispatch_async(mQueue, mSubscriptionEstablishedHandler);
+        auto subscriptionEstablishedHandler = mSubscriptionEstablishedHandler;
+        subscriptionEstablishedHandler();
     }
 }
 
@@ -109,9 +106,7 @@
         auto callback = mResubscriptionCallback;
         auto error = [MTRError errorForCHIPErrorCode:aTerminationCause];
         auto delayMs = @(apReadClient->ComputeTimeTillNextSubscription());
-        dispatch_async(mQueue, ^{
-            callback(error, delayMs);
-        });
+        callback(error, delayMs);
     }
     return CHIP_NO_ERROR;
 }
@@ -129,30 +124,21 @@
         return;
     }
 
-    __block ErrorCallback callback = mErrorCallback;
     __block auto * myself = this;
+
+    auto errorCallback = mErrorCallback;
     mErrorCallback = nil;
     mAttributeReportCallback = nil;
     mEventReportCallback = nil;
-    __auto_type onDoneHandler = mOnDoneHandler;
+    auto onDoneHandler = mOnDoneHandler;
     mOnDoneHandler = nil;
-    dispatch_async(mQueue, ^{
-        callback(err);
-    });
 
+    errorCallback(err);
     if (onDoneHandler) {
-        // To guarantee the async onDoneHandler call is made before
-        // deletion, so that clean up can happen while the callback
-        // object is still alive (and therefore cluster cache), queue
-        // deletion after calling the onDoneHandler
-        mHaveQueuedDeletion = true;
-        dispatch_async(mQueue, ^{
-            onDoneHandler();
-            dispatch_async(DeviceLayer::PlatformMgrImpl().GetWorkQueue(), ^{
-                delete myself;
-            });
-        });
-    } else if (aCancelSubscription) {
+        onDoneHandler();
+    }
+
+    if (aCancelSubscription) {
         // We can't synchronously delete ourselves, because we're inside one of
         // the ReadClient callbacks and we need to outlive the callback's
         // execution.  Queue an async deletion on the Matter queue (where we are