Add option to mark CONNECTED only on initial sync

Summary:
In Dsf we use SubscriptionState::CONNECTED to determine that we actually have all the state we need.

Previously, fsdb would serve heartbeats after serving initial sync so first chunk revieved would always be initial sync and we'd go into CONNECTED. But now heartbeats are served on a dedicated thread to avoid starvation, so we need to make sure the client doesnt prematurely mark CONNECTED.

Ideally we can always follow this behavior but actually switch agent relies on the reverse (CONNECTED to mean any chunk came through). For now introducing a flag to allow both behaviors, perhaps we can introduce a new state SYNCED to differentiate

Differential Revision: D63554533

fbshipit-source-id: d260d48b1d7dea4b02a7c5caed6fd77d11222456

fboss/fsdb/client/FsdbDeltaSubscriber.cpp

@@ -44,13 +44,17 @@ FsdbDeltaSubscriberImpl<SubUnit, PathElement>::serveStream(StreamT&& stream) {
       XLOG(DBG2) << " Detected cancellation: " << this->clientId();
       break;
     }
-    // even empty change/heartbeat indicates subscription is connected
-    if (this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+    if (!this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+        this->getSubscriptionState() != SubscriptionState::CONNECTED) {
       BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
     }
     if (!delta->changes()->size()) {
       continue;
     }
+    if (this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+        this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+      BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
+    }
     SubUnitT tmp(*delta);
     this->operSubUnitUpdate_(std::move(tmp));
   }

fboss/fsdb/client/FsdbPatchSubscriber.cpp

@@ -47,12 +47,16 @@ FsdbPatchSubscriberImpl<MessageType, SubUnit, PathElement>::serveStream(
       XLOG(DBG2) << " Detected cancellation: " << this->clientId();
       break;
     }
-    // even empty change/heartbeat indicates subscription is connected
-    if (this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+    if (!this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+        this->getSubscriptionState() != SubscriptionState::CONNECTED) {
       BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
     }
     switch (message->getType()) {
       case SubscriberMessage::Type::chunk:
+        if (this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+            this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+          BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
+        }
         this->operSubUnitUpdate_(message->move_chunk());
         break;
       case SubscriberMessage::Type::heartbeat:

fboss/fsdb/client/FsdbStateSubscriber.cpp

@@ -45,8 +45,8 @@ FsdbStateSubscriberImpl<SubUnit, PathElement>::serveStream(StreamT&& stream) {
       XLOG(DBG2) << " Detected cancellation: " << this->clientId();
       break;
     }
-    // even empty change/heartbeat indicates subscription is connected
-    if (this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+    if (!this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+        this->getSubscriptionState() != SubscriptionState::CONNECTED) {
       BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
     }
     if constexpr (std::is_same_v<SubUnitT, OperState>) {
@@ -58,6 +58,10 @@ FsdbStateSubscriberImpl<SubUnit, PathElement>::serveStream(StreamT&& stream) {
         continue;
       }
     }
+    if (this->subscriptionOptions().requireInitialSyncToMarkConnect_ &&
+        this->getSubscriptionState() != SubscriptionState::CONNECTED) {
+      BaseT::updateSubscriptionState(SubscriptionState::CONNECTED);
+    }
     SubUnitT tmp(*state);
     this->operSubUnitUpdate_(std::move(tmp));
   }

fboss/fsdb/client/FsdbSubscriber.h

@@ -83,14 +83,18 @@ struct SubscriptionOptions {
   explicit SubscriptionOptions(
       const std::string& clientId,
       bool subscribeStats = false,
-      uint32_t grHoldTimeSec = 0)
+      uint32_t grHoldTimeSec = 0,
+      // only mark subscription as CONNECTED on initial sync
+      bool requireInitialSyncToMarkConnect = false)
       : clientId_(clientId),
         subscribeStats_(subscribeStats),
-        grHoldTimeSec_(grHoldTimeSec) {}
+        grHoldTimeSec_(grHoldTimeSec),
+        requireInitialSyncToMarkConnect_(requireInitialSyncToMarkConnect) {}
 
   const std::string clientId_;
   bool subscribeStats_{false};
   uint32_t grHoldTimeSec_{0};
+  bool requireInitialSyncToMarkConnect_{false};
 };
 
 struct SubscriptionInfo {
@@ -283,6 +287,10 @@ class FsdbSubscriber : public FsdbSubscriberBase {
     return subscribePaths_;
   }
 
+  const SubscriptionOptions& subscriptionOptions() const {
+    return subscriptionOptions_;
+  }
+
   FsdbSubUnitUpdateCb operSubUnitUpdate_;
 
  private: