Merge 932f91c into fa450dd

Author: snaury

ydb/core/tx/datashard/datashard.cpp

@@ -3682,12 +3682,13 @@ void TDataShard::Handle(TEvMediatorTimecast::TEvSubscribeReadStepResult::TPtr& e
     auto it = CoordinatorSubscriptionById.find(msg->CoordinatorId);
     Y_VERIFY_S(it != CoordinatorSubscriptionById.end(),
         "Unexpected TEvSubscribeReadStepResult for coordinator " << msg->CoordinatorId);
-    size_t index = it->second;
-    auto& subscription = CoordinatorSubscriptions.at(index);
-    subscription.ReadStep = msg->ReadStep;
     CoordinatorPrevReadStepMin = Max(CoordinatorPrevReadStepMin, msg->LastReadStep);
     CoordinatorPrevReadStepMax = Min(CoordinatorPrevReadStepMax, msg->NextReadStep);
     --CoordinatorSubscriptionsPending;
+
+    // Note: we don't use the subscription and unsubscribe immediately
+    ctx.Send(MakeMediatorTimecastProxyID(), new TEvMediatorTimecast::TEvUnsubscribeReadStep(msg->CoordinatorId));
+
     CheckMediatorStateRestored();
 }
 

ydb/core/tx/datashard/datashard_impl.h

@@ -2703,7 +2703,6 @@ class TDataShard
 
     struct TCoordinatorSubscription {
         ui64 CoordinatorId;
-        TMediatorTimecastReadStep::TCPtr ReadStep;
     };
 
     TVector<TCoordinatorSubscription> CoordinatorSubscriptions;

ydb/core/tx/time_cast/time_cast.cpp

@@ -17,9 +17,6 @@
 
 namespace NKikimr {
 
-// We will unsubscribe from idle coordinators after 5 minutes
-static constexpr TDuration MaxIdleCoordinatorSubscriptionTime = TDuration::Minutes(5);
-
 ui64 TMediatorTimecastEntry::Get(ui64 tabletId) const {
     Y_UNUSED(tabletId);
     return AtomicGet(Step);
@@ -115,7 +112,7 @@ class TMediatorTimecastProxy : public TActor<TMediatorTimecastProxy> {
         THashSet<TActorId> Subscribers;
         TMap<std::pair<ui64, TActorId>, ui64> SubscribeRequests; // (seqno, subscriber) -> Cookie
         TMap<std::pair<ui64, TActorId>, ui64> WaitRequests; // (step, subscriber) -> Cookie
-        TMonotonic IdleStart;
+        bool Unsubscribed = false;
     };
 
     THashMap<ui64, TMediator> Mediators; // mediator tablet -> info
@@ -190,6 +187,7 @@ class TMediatorTimecastProxy : public TActor<TMediatorTimecastProxy> {
 
     void SyncCoordinator(ui64 coordinatorId, TCoordinator &coordinator, const TActorContext &ctx);
     void ResyncCoordinator(ui64 coordinatorId, const TActorId &pipeClient, const TActorContext &ctx);
+    void UnsubscribeCoordinator(ui64 coordinatorId, TCoordinator &coordinator, const TActorContext &ctx);
     void NotifyCoordinatorWaiters(ui64 coordinatorId, TCoordinator &coordinator, const TActorContext &ctx);
 
     void Handle(TEvMediatorTimecast::TEvRegisterTablet::TPtr &ev, const TActorContext &ctx);
@@ -442,8 +440,9 @@ void TMediatorTimecastProxy::SyncCoordinator(ui64 coordinatorId, TCoordinator &c
             NTabletPipe::CreateClient(ctx.SelfID, coordinatorId, retryPolicy));
     }
 
-    coordinator.LastSentSeqNo = seqNo;
     NTabletPipe::SendData(ctx, coordinator.PipeClient, new TEvTxProxy::TEvSubscribeReadStep(coordinatorId, seqNo));
+    coordinator.LastSentSeqNo = seqNo;
+    coordinator.Unsubscribed = false;
 }
 
 void TMediatorTimecastProxy::ResyncCoordinator(ui64 coordinatorId, const TActorId &pipeClient, const TActorContext &ctx) {
@@ -468,6 +467,17 @@ void TMediatorTimecastProxy::ResyncCoordinator(ui64 coordinatorId, const TActorI
     SyncCoordinator(coordinatorId, coordinator, ctx);
 }
 
+void TMediatorTimecastProxy::UnsubscribeCoordinator(ui64 coordinatorId, TCoordinator &coordinator, const TActorContext &ctx) {
+    if (coordinator.Unsubscribed || !coordinator.PipeClient) {
+        return;
+    }
+
+    // Note: we leave the pipe open to make new subscriptions faster
+    // The idle entry will be removed when the pipe eventually disconnects
+    NTabletPipe::SendData(ctx, coordinator.PipeClient, new TEvTxProxy::TEvUnsubscribeReadStep(coordinatorId, coordinator.LastSentSeqNo));
+    coordinator.Unsubscribed = true;
+}
+
 void TMediatorTimecastProxy::Handle(TEvMediatorTimecast::TEvSubscribeReadStep::TPtr &ev, const TActorContext &ctx) {
     const auto *msg = ev->Get();
     LOG_DEBUG_S(ctx, NKikimrServices::TX_MEDIATOR_TIMECAST, "Actor# " << ctx.SelfID
@@ -496,7 +506,7 @@ void TMediatorTimecastProxy::Handle(TEvMediatorTimecast::TEvUnsubscribeReadStep:
             auto &coordinator = Coordinators[coordinatorId];
             coordinator.Subscribers.erase(ev->Sender);
             if (coordinator.Subscribers.empty()) {
-                coordinator.IdleStart = ctx.Monotonic();
+                UnsubscribeCoordinator(coordinatorId, coordinator, ctx);
             }
         }
         subscriber.Coordinators.clear();
@@ -505,7 +515,7 @@ void TMediatorTimecastProxy::Handle(TEvMediatorTimecast::TEvUnsubscribeReadStep:
         auto &coordinator = Coordinators[msg->CoordinatorId];
         coordinator.Subscribers.erase(ev->Sender);
         if (coordinator.Subscribers.empty()) {
-            coordinator.IdleStart = ctx.Monotonic();
+            UnsubscribeCoordinator(msg->CoordinatorId, coordinator, ctx);
         }
         subscriber.Coordinators.erase(msg->CoordinatorId);
     }
@@ -614,16 +624,6 @@ void TMediatorTimecastProxy::Handle(TEvTxProxy::TEvSubscribeReadStepUpdate::TPtr
 
         NotifyCoordinatorWaiters(coordinatorId, coordinator, ctx);
     }
-
-    // Unsubscribe from idle coordinators
-    if (coordinator.Subscribers.empty() && (ctx.Monotonic() - coordinator.IdleStart) >= MaxIdleCoordinatorSubscriptionTime) {
-        if (coordinator.PipeClient) {
-            NTabletPipe::CloseClient(ctx, coordinator.PipeClient);
-            coordinator.PipeClient = TActorId();
-        }
-
-        Coordinators.erase(itCoordinator);
-    }
 }
 
 void TMediatorTimecastProxy::NotifyCoordinatorWaiters(ui64 coordinatorId, TCoordinator &coordinator, const TActorContext &ctx) {