Fix after review

Author: pixcc

ydb/core/mind/node_broker.cpp

@@ -687,7 +687,12 @@ void TNodeBroker::SubscribeForConfigUpdates(const TActorContext &ctx)
 
 void TNodeBroker::SendToSubscriber(const TSubscriberInfo &subscriber, IEventBase* event, const TActorContext &ctx) const
 {
-    THolder<IEventHandle> ev = MakeHolder<IEventHandle>(subscriber.Id, ctx.SelfID, event);
+    SendToSubscriber(subscriber, event, 0, ctx);
+}
+
+void TNodeBroker::SendToSubscriber(const TSubscriberInfo &subscriber, IEventBase* event, ui64 cookie, const TActorContext &ctx) const
+{
+    THolder<IEventHandle> ev = MakeHolder<IEventHandle>(subscriber.Id, ctx.SelfID, event, 0, cookie);
     if (subscriber.PipeServerInfo->IcSession) {
         ev->Rewrite(TEvInterconnect::EvForward, subscriber.PipeServerInfo->IcSession);
     }
@@ -708,10 +713,6 @@ void TNodeBroker::SendUpdateNodes(const TActorContext &ctx)
 
 void TNodeBroker::SendUpdateNodes(const TSubscriberInfo &subscriber, ui64 version, const TActorContext &ctx)
 {
-    if (version >= Committed.Epoch.Version) {
-        return;
-    }
-
     NKikimrNodeBroker::TUpdateNodes record;
     record.SetSeqNo(subscriber.SeqNo);
     Committed.Epoch.Serialize(*record.MutableEpoch());
@@ -743,15 +744,15 @@ TNodeBroker::TSubscriberInfo& TNodeBroker::AddSubscriber(TActorId subscriberId,
 
     auto& pipeServer = PipeServers.at(pipeServerId);
     auto res = Subscribers.emplace(subscriberId, TSubscriberInfo(subscriberId, seqNo, &pipeServer));
-    Y_VERIFY_DEBUG_S(res.second, "Subscription already exists for " << subscriberId);
+    Y_ENSURE(res.second, "Subscription already exists for " << subscriberId);
     pipeServer.Subscribers.insert(subscriberId);
     return res.first->second;
 }
 
 void TNodeBroker::RemoveSubscriber(TActorId subscriber, const TActorContext &ctx)
 {
     auto it = Subscribers.find(subscriber);
-    Y_VERIFY_DEBUG_S(it != Subscribers.end(), "No subscription for " << subscriber);
+    Y_ENSURE(it != Subscribers.end(), "No subscription for " << subscriber);
 
     LOG_DEBUG_S(ctx, NKikimrServices::NODE_BROKER,
                 "Unsubscribed " << subscriber
@@ -1622,26 +1623,25 @@ void TNodeBroker::Handle(TEvNodeBroker::TEvSyncNodesRequest::TPtr &ev,
     TabletCounters->Cumulative()[COUNTER_SYNC_NODES_REQUESTS].Increment(1);
 
     if (auto it = Subscribers.find(ev->Sender); it != Subscribers.end()) {
-        auto response = MakeHolder<TEvNodeBroker::TEvSyncNodesResponse>();
-        response->Record.SetSeqNo(it->second.SeqNo);
-        SendToSubscriber(it->second, response.Release(), ctx);
-    } else {
-        LOG_ERROR_S(ctx, NKikimrServices::NODE_BROKER,
-                    "Unexpected TEvSyncNodesRequest without subscription from " << ev->Sender);
+        if (it->second.SeqNo == ev->Get()->Record.GetSeqNo()) {
+            auto response = MakeHolder<TEvNodeBroker::TEvSyncNodesResponse>();
+            response->Record.SetSeqNo(it->second.SeqNo);
+            SendToSubscriber(it->second, response.Release(), ev->Cookie, ctx);
+        }
     }
 }
 
 void TNodeBroker::Handle(TEvTabletPipe::TEvServerConnected::TPtr &ev)
 {
     auto res = PipeServers.emplace(ev->Get()->ServerId, TPipeServerInfo(ev->Get()->ServerId, ev->Get()->InterconnectSession));
-    Y_VERIFY_DEBUG_S(res.second, "Unexpected TEvServerConnected for " << ev->Get()->ServerId);
+    Y_ENSURE(res.second, "Unexpected TEvServerConnected for " << ev->Get()->ServerId);
 }
 
 void TNodeBroker::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr &ev,
                          const TActorContext &ctx)
 {
     auto it = PipeServers.find(ev->Get()->ServerId);
-    Y_VERIFY_DEBUG_S(it != PipeServers.end(), "Unexpected TEvServerDisconnected for " << ev->Get()->ServerId);
+    Y_ENSURE(it != PipeServers.end(), "Unexpected TEvServerDisconnected for " << ev->Get()->ServerId);
     while (!it->second.Subscribers.empty()) {
         RemoveSubscriber(*it->second.Subscribers.begin(), ctx);
     }

ydb/core/mind/node_broker__migrate_state.cpp

@@ -102,7 +102,10 @@ class TNodeBroker::TTxMigrateState : public TTransactionBase<TNodeBroker> {
             Self->ScheduleEpochUpdate(ctx);
             Self->PrepareEpochCache();
             Self->PrepareUpdateNodesLog();
-            Self->SignalTabletActive(ctx, "1.0");
+
+            NKikimrNodeBroker::TVersionInfo versionInfo;
+            versionInfo.SetSupportDeltaProtocol(true);
+            Self->SignalTabletActive(ctx, versionInfo.SerializeAsString());
         } else {
             Self->Execute(Self->CreateTxMigrateState(std::move(DbChanges)));
         }

ydb/core/mind/node_broker_impl.h

@@ -286,6 +286,7 @@ class TNodeBroker : public TActor<TNodeBroker>
     void SendUpdateNodes(const TActorContext &ctx);
     void SendUpdateNodes(const TSubscriberInfo &subscriber, ui64 version, const TActorContext &ctx);
     void SendToSubscriber(const TSubscriberInfo &subscriber, IEventBase* event, const TActorContext &ctx) const;
+    void SendToSubscriber(const TSubscriberInfo &subscriber, IEventBase* event, ui64 cookie, const TActorContext &ctx) const;
 
     TSubscriberInfo& AddSubscriber(TActorId subscriberId, TActorId pipeServerId, ui64 seqNo, const TActorContext &ctx);
     void RemoveSubscriber(TActorId subscriber, const TActorContext &ctx);

ydb/core/mind/node_broker_ut.cpp

@@ -842,13 +842,7 @@ void CheckUpdateNodesLog(TTestActorRuntime &runtime,
     static ui32 seqNo = 1;
     TActorId pipe = runtime.ConnectToPipe(MakeNodeBrokerID(), sender, 0, GetPipeConfigWithRetries());
     SubscribeToNodesUpdates(runtime, sender, pipe, version, ++seqNo);
-    if (epoch.GetVersion() != version) {
-        CheckNodesUpdate(runtime, updates, seqNo, epoch);
-    } else {
-        TBlockEvents<TEvNodeBroker::TEvUpdateNodes> block(runtime);
-        runtime.SimulateSleep(TDuration::MilliSeconds(10));
-        UNIT_ASSERT(block.empty());
-    }
+    CheckNodesUpdate(runtime, updates, seqNo, epoch);
     runtime.ClosePipe(pipe, sender, 0);
 }
 

ydb/core/protos/node_broker.proto

@@ -37,6 +37,10 @@ message TNodeInfoSchema {
     optional bool AuthorizedByCertificate = 10;
 }
 
+message TVersionInfo {
+    optional bool SupportDeltaProtocol = 1;
+}
+
 message TEpoch {
     optional uint64 Id = 1;
     optional uint64 Version = 2;
@@ -164,6 +168,7 @@ message TUpdateNodes {
 }
 
 message TSyncNodesRequest {
+    optional uint64 SeqNo = 1;
 }
 
 message TSyncNodesResponse {