Add senders readiness helpers (#9103)

Author: Enjection

ydb/core/change_exchange/change_sender_common_ops.h

@@ -283,6 +283,7 @@ class TBaseChangeSender {
 
         Y_ABORT_UNLESS(sender.Ready);
         sender.Ready = false;
+        ReadySenders--;
 
         sender.Pending.reserve(sender.Prepared.size());
         for (const auto& record : sender.Prepared) {
@@ -438,6 +439,7 @@ class TBaseChangeSender {
                 ActorOps->Send(sender.ActorId, new TEvents::TEvPoisonPill());
             }
         }
+        ReadySenders = 0;
     }
 
     void RemoveRecords() {
@@ -531,6 +533,7 @@ class TBaseChangeSender {
 
         auto& sender = it->second;
         sender.Ready = true;
+        ReadySenders++;
 
         if (sender.Pending) {
             RemoveRecords(std::exchange(sender.Pending, {}));
@@ -555,6 +558,9 @@ class TBaseChangeSender {
         }
 
         ReEnqueueRecords(it->second);
+        if (it->second.Ready) {
+            --ReadySenders;
+        }
         Senders.erase(it);
         GonePartitions.push_back(partitionId);
 
@@ -580,6 +586,10 @@ class TBaseChangeSender {
             , MemUsage(0)
     {}
 
+    bool IsAllSendersReady() {
+        return ReadySenders == Senders.size();
+    }
+
     void RenderHtmlPage(ui64 tabletId, NMon::TEvRemoteHttpInfo::TPtr& ev, const TActorContext& ctx) {
         const auto& cgi = ev->Get()->Cgi();
         if (const auto& str = cgi.Get("partitionId")) {
@@ -782,6 +792,7 @@ class TBaseChangeSender {
     ui64 MemUsage;
 
     THashMap<ui64, TSender> Senders; // ui64 is partition id
+    ui64 ReadySenders = 0;
     TSet<TEnqueuedRecord> Enqueued;
     TSet<TIncompleteRecord> PendingBody;
     TMap<ui64, typename TChangeRecord::TPtr> PendingSent; // ui64 is order