Simplify Channel 2.0 FSM, eliminate extra doubles, fix some bugs (#44864)

4 files changed, 395 insertions, 337 deletions

diff --git a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
index 2935c169ebe..e7b464503f5 100644
--- a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
+++ b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
@@ -350,7 +350,7 @@ public:
             limits.RemoteChannelInflightBytes = config.GetRemoteChannelInflightBytes();
             limits.RemoteSessionInflightBytes = config.GetRemoteSessionInflightBytes();
             limits.ReconciliationCount = config.GetReconciliationCount();
-            limits.CleanupPeriod = TDuration::MilliSeconds(std::max<ui64>(config.GetCleanupPeriodMs(), 200));
+            limits.CleanupPeriod = TDuration::MilliSeconds(config.GetCleanupPeriodMs());
             limits.IdlePingPeriod = TDuration::MilliSeconds(config.GetIdlePingPeriodMs());
             limits.IdleDestroyPeriod = TDuration::MilliSeconds(config.GetIdleDestroyPeriodMs());
         } else { // deprecated
diff --git a/ydb/core/kqp/ut/channels/dq_channel_service_ut.cpp b/ydb/core/kqp/ut/channels/dq_channel_service_ut.cpp
index d6935ee8eaa..1c272f4dbc6 100644
--- a/ydb/core/kqp/ut/channels/dq_channel_service_ut.cpp
+++ b/ydb/core/kqp/ut/channels/dq_channel_service_ut.cpp
@@ -58,6 +58,8 @@ struct TWorkerSettings {
     int MinMessageSize = 10;
     int MaxMessageSize = 10000;
     bool EarlyFinish = false;
+    int PauseMessageIndex = -1;
+    int PauseDelayMs = 0;
 };
 
 struct TFailureSettings {
@@ -67,12 +69,13 @@ struct TFailureSettings {
     int Discovery = 0;
 };
 
-class TWorkerActor : public NActors::TActor<TWorkerActor> {
+template <typename TDerived>
+class TWorkerActor : public NActors::TActor<TDerived> {
 public:
-    TWorkerActor(std::shared_ptr<IDqChannelService> service, TEvTestPrivate::ERole role, ui32 channelId, const TWorkerSettings& settings)
-        : TActor(&TThis::StateFunc)
+    TWorkerActor(const TString& logPrefix, std::shared_ptr<IDqChannelService> service, ui32 channelId, const TWorkerSettings& settings)
+        : NActors::TActor<TDerived>(&TWorkerActor::StateFunc)
+        , LogPrefix(logPrefix)
         , Service(service)
-        , Role(role)
         , ChannelId(channelId)
         , Settings(settings)
     {}
@@ -86,98 +89,38 @@ public:
         }
     }
 
-    void Run() {
-        if (!Started) {
-            switch (Role) {
-                case TEvTestPrivate::ERole::Producer: {
-                    TChannelFullInfo info(ChannelId, SelfId(), PeerId, 0, 1, TCollectStatsLevel::None);
-                    Buffer = Service->GetOutputBuffer(info, nullptr, nullptr);
-                    break;
-                }
-                case TEvTestPrivate::ERole::Consumer: {
-                    TChannelFullInfo info(ChannelId, PeerId, SelfId(), 0, 1, TCollectStatsLevel::None);
-                    Buffer = Service->GetInputBuffer(info, nullptr);
-                    break;
-                }
-            }
-            Started = true;
-        }
-
-        switch (Role) {
-            case TEvTestPrivate::ERole::Producer: {
-                if (Buffer->IsFinished()) {
-                    LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, "TEST FINISHED SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
-                    Send(RunnerId, new TEvTestPrivate::TEvFinished(Role, false));
-                    PassAway();
-                    return;
-                }
-                while (Buffer->GetFillLevel() == EDqFillLevel::NoLimit && MessageIndex <= Settings.MessageCount) {
-                    if (MessageIndex == Settings.MessageCount) {
-                        Buffer->SendFinish();
-                    } else {
-                        auto bytes = Settings.MinMessageSize + RandomNumber<ui64>(Settings.MaxMessageSize - Settings.MinMessageSize);
-                        Buffer->Push(TDataChunk(NYql::TChunkedBuffer(TString(bytes, 'a')), 1, false));
-                    }
-                    MessageIndex++;
-                }
-                break;
-            }
-            case TEvTestPrivate::ERole::Consumer: {
-                TDataChunk data;
-                while (MessageIndex < Settings.MessageCount && Buffer->Pop(data)) {
-                    MessageIndex++;
-                }
-                if (Settings.EarlyFinish && MessageIndex == Settings.MessageCount) {
-                    LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, "TEST EARLY FINISH SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
-                    Buffer->EarlyFinish();
-                    MessageIndex++;
-                }
-                if (MessageIndex <= Settings.MessageCount && Buffer->Pop(data)) {
-                    MessageIndex++;
-                }
-                if (Buffer->IsFinished()) {
-                    LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, "TEST FINISHED SelfId=" << SelfId() << ", ChannelId=" << ChannelId
-                        << ", Role=" << (Role == TEvTestPrivate::ERole::Producer ? "Producer" : "Consumer"));
-                    Send(RunnerId, new TEvTestPrivate::TEvFinished(Role, false));
-                    PassAway();
-                }
-                break;
-            }
-        }
-    }
+    virtual void Run() = 0;
 
-    void HandleWakeup(NActors::TEvents::TEvWakeup::TPtr&) {
+    virtual void HandleWakeup(NActors::TEvents::TEvWakeup::TPtr&) {
         Run();
     }
 
-    void HandleResume(TEvDqCompute::TEvResumeExecution::TPtr&) {
+    virtual void HandleResume(TEvDqCompute::TEvResumeExecution::TPtr&) {
         Run();
     }
 
-    void HandleStart(TEvTestPrivate::TEvStart::TPtr& ev) {
+    virtual void HandleStart(TEvTestPrivate::TEvStart::TPtr& ev) {
         RunnerId = ev->Sender;
         PeerId = ev->Get()->PeerId;
-        LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS,
-            "TEST WORKER START SelfId=" << SelfId() << ", ChannelId=" << ChannelId << ", PeerId=" << PeerId << ", Role=" << (Role == TEvTestPrivate::ERole::Producer ? "Producer" : "Consumer")
-        );
+        LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST START SelfId=" << this->SelfId() << ", ChannelId=" << ChannelId << ", PeerId=" << PeerId);
         if (Settings.StartDelayMs) {
-            Schedule(TDuration::MilliSeconds(RandomNumber<ui64>(Settings.StartDelayMs) + 1), new NActors::TEvents::TEvWakeup());
+            this->Schedule(TDuration::MilliSeconds(RandomNumber<ui64>(Settings.StartDelayMs) + 1), new NActors::TEvents::TEvWakeup());
         } else {
             Run();
         }
     }
 
-    void HandleAbort(NYql::NDq::TEvDq::TEvAbortExecution::TPtr& ev) {
+    virtual void HandleAbort(NYql::NDq::TEvDq::TEvAbortExecution::TPtr& ev) {
         LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS,
-            "TEST WORKER ABORT SelfId=" << SelfId() << ", ChannelId=" << ChannelId << ", " << ev->Get()->GetIssues().ToOneLineString()
+            LogPrefix << "TEST ABORT SelfId=" << this->SelfId() << ", ChannelId=" << ChannelId << ", " << ev->Get()->GetIssues().ToOneLineString()
         );
-        Send(RunnerId, new TEvTestPrivate::TEvFinished(Role, true));
-        PassAway();
+        this->Send(RunnerId, new TEvTestPrivate::TEvFinished(TEvTestPrivate::ERole::Producer, true));
+        this->PassAway();
     }
 
+    TString LogPrefix;
     std::shared_ptr<IDqChannelService> Service;
     std::shared_ptr<IChannelBuffer> Buffer;
-    TEvTestPrivate::ERole Role;
     ui32 ChannelId;
     NActors::TActorId PeerId;
     NActors::TActorId RunnerId;
@@ -186,105 +129,246 @@ public:
     bool Started = false;
 };
 
-Y_UNIT_TEST_SUITE(Channels20) {
-
-    void LoadTest(int count, bool local, const TWorkerSettings& producerSettings, const TWorkerSettings& consumerSettings, const TFailureSettings& failureSettings = TFailureSettings{}) {
+class TProducerActor : public TWorkerActor<TProducerActor> {
+public:
+    TProducerActor(std::shared_ptr<IDqChannelService> service, ui32 channelId, const TWorkerSettings& settings)
+        : TWorkerActor("PROD ", service, channelId, settings)
+    {}
 
-        TKikimrSettings settings;
-        settings.NodeCount = local ? 1 : 2;
-        settings.LogSettings = TTestLogSettings().AddLogPriority(NKikimrServices::KQP_CHANNELS, NActors::NLog::EPriority::PRI_TRACE);
-        settings.LogSettings->DefaultLogPriority = NActors::NLog::EPriority::PRI_CRIT;
-        TKikimrRunner kikimr(settings);
-        auto& runtime = *kikimr.GetTestServer().GetRuntime();
-        runtime.SetUseRealInterconnect();
+    void Run() override {
+        if (!Started) {
+            TChannelFullInfo info(ChannelId, SelfId(), PeerId, 0, 1, TCollectStatsLevel::None);
+            Buffer = Service->GetOutputBuffer(info, nullptr, nullptr);
+            Started = true;
+        }
+        if (Buffer->IsFinished()) {
+            LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST FINISHED SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
+            Send(RunnerId, new TEvTestPrivate::TEvFinished(TEvTestPrivate::ERole::Producer, false));
+            PassAway();
+            return;
+        }
+        while (Buffer->GetFillLevel() == EDqFillLevel::NoLimit && MessageIndex < Settings.MessageCount) {
+            if (Settings.PauseMessageIndex == MessageIndex) {
+                if (!ResumeTime) {
+                    ResumeTime = TInstant::Now() + TDuration::MilliSeconds(Settings.PauseDelayMs);
+                    LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST PAUSED SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
+                }
+                if (TInstant::Now() < ResumeTime) {
+                    Schedule(ResumeTime, new NActors::TEvents::TEvWakeup());
+                    return;
+                } else {
+                    ResumeTime = TInstant::Zero();
+                    LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST RESUMED SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
+                }
+            }
+            auto bytes = Settings.MinMessageSize + RandomNumber<ui64>(Settings.MaxMessageSize - Settings.MinMessageSize);
+            Buffer->Push(TDataChunk(NYql::TChunkedBuffer(TString(bytes, 'a')), 1, false));
+            MessageIndex++;
+        }
+        if (MessageIndex == Settings.MessageCount) {
+            Buffer->SendFinish();
+        }
+    }
 
-        auto control0 = runtime.AllocateEdgeActor(0);
-        auto control1 = local ? control0 : runtime.AllocateEdgeActor(1);
+    TInstant ResumeTime;
+};
 
-        std::shared_ptr<TDqChannelService> service0;
-        std::shared_ptr<TDqChannelService> service1;
+class TConsumerActor : public TWorkerActor<TConsumerActor> {
+public:
+    TConsumerActor(std::shared_ptr<IDqChannelService> service, ui32 channelId, const TWorkerSettings& settings)
+        : TWorkerActor("CONS ", service, channelId, settings)
+    {}
 
-        ui32 nodeIndex0 = 0;
-        ui32 nodeIndex1 = local ? 0 : 1;
+    void Run() override {
+        if (!Started) {
+            TChannelFullInfo info(ChannelId, PeerId, SelfId(), 0, 1, TCollectStatsLevel::None);
+            Buffer = Service->GetInputBuffer(info, nullptr);
+            Started = true;
+        }
+        TDataChunk data;
+        while (MessageIndex < Settings.MessageCount && Buffer->Pop(data)) {
+            MessageIndex++;
+        }
+        if (Settings.EarlyFinish && MessageIndex == Settings.MessageCount) {
+            LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST EARLY FINISH SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
+            Buffer->EarlyFinish();
+            MessageIndex++;
+        }
+        if (MessageIndex <= Settings.MessageCount && Buffer->Pop(data)) {
+            MessageIndex++;
+        }
+        if (Buffer->IsFinished()) {
+            LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_CHANNELS, LogPrefix << "TEST FINISHED SelfId=" << SelfId() << ", ChannelId=" << ChannelId);
+            Send(RunnerId, new TEvTestPrivate::TEvFinished(TEvTestPrivate::ERole::Consumer, false));
+            PassAway();
+        }
+    }
+};
 
-        runtime.Send(MakeChannelServiceActorID(runtime.GetNodeId(0)), control0, new TEvPrivate::TEvServiceLookup(), nodeIndex0);
-        auto serviceReply = runtime.GrabEdgeEvent<TEvPrivate::TEvServiceReply>(control0)->Release();
-        service0 = serviceReply->Service;
+struct TLoadTest {
 
-        if (local) {
-            service1 = service0;
-        } else {
-            runtime.Send(MakeChannelServiceActorID(runtime.GetNodeId(1)), control1, new TEvPrivate::TEvServiceLookup(), nodeIndex1);
-            auto serviceReply = runtime.GrabEdgeEvent<TEvPrivate::TEvServiceReply>(control1)->Release();
-            service1 = serviceReply->Service;
+    virtual void Prepare() {
+        settings.NodeCount = Local ? 1 : 2;
+        settings.LogSettings = TTestLogSettings().AddLogPriority(NKikimrServices::KQP_CHANNELS, NActors::NLog::EPriority::PRI_TRACE);
+        settings.LogSettings->DefaultLogPriority = NActors::NLog::EPriority::PRI_CRIT;
+        if (Local) {
+            NodeIndex1 = NodeIndex0;
         }
+    }
 
-        THashSet<NActors::TActorId> actors;
+    virtual void Init() {
+        Runner = std::make_unique<TKikimrRunner>(settings);
+        Runtime = Runner->GetTestServer().GetRuntime();
+        Runtime->SetUseRealInterconnect();
 
-        std::atomic<int> dataCount = 0;
-        NActors::TTestActorRuntime::TEventObserverHolder dataHolder;
-        if (failureSettings.Data) {
-            dataHolder = runtime.AddObserver<TEvDqCompute::TEvChannelDataV2>([&](TEvDqCompute::TEvChannelDataV2::TPtr& event) {
-                if (dataCount.fetch_add(1) == failureSettings.Data) {
-                    dataCount.fetch_sub(failureSettings.Data);
-                    event.Reset();
-                }
-            });
+        Control0 = Runtime->AllocateEdgeActor(0);
+        Control1 = Local ? Control0 : Runtime->AllocateEdgeActor(1);
+
+        Runtime->Send(MakeChannelServiceActorID(Runtime->GetNodeId(0)), Control0, new TEvPrivate::TEvServiceLookup(), NodeIndex0);
+        auto serviceReply = Runtime->GrabEdgeEvent<TEvPrivate::TEvServiceReply>(Control0)->Release();
+        Service0 = serviceReply->Service;
+
+        if (Local) {
+            Service1 = Service0;
+        } else {
+            Runtime->Send(MakeChannelServiceActorID(Runtime->GetNodeId(1)), Control1, new TEvPrivate::TEvServiceLookup(), NodeIndex1);
+            auto serviceReply = Runtime->GrabEdgeEvent<TEvPrivate::TEvServiceReply>(Control1)->Release();
+            Service1 = serviceReply->Service;
         }
+    }
 
-        for (auto i = 0; i < count; i ++) {
+    virtual void Start() {
+        for (auto i = 0; i < Count; i ++) {
             auto channelId = i + 1;
             if ((i & 1) == 0) {
-                auto producer = runtime.Register(new TWorkerActor(service0, TEvTestPrivate::ERole::Producer, channelId, producerSettings), nodeIndex0);
-                auto consumer = runtime.Register(new TWorkerActor(service1, TEvTestPrivate::ERole::Consumer, channelId, consumerSettings), nodeIndex1);
-                runtime.Send(consumer, control1, new TEvTestPrivate::TEvStart(producer), nodeIndex1, true);
-                runtime.Send(producer, control0, new TEvTestPrivate::TEvStart(consumer), nodeIndex0, true);
-                actors.insert(producer);
-                actors.insert(consumer);
+                auto producer = Runtime->Register(new TProducerActor(Service0, channelId, ProducerSettings), NodeIndex0);
+                auto consumer = Runtime->Register(new TConsumerActor(Service1, channelId, ConsumerSettings), NodeIndex1);
+                Runtime->Send(consumer, Control1, new TEvTestPrivate::TEvStart(producer), NodeIndex1, true);
+                Runtime->Send(producer, Control0, new TEvTestPrivate::TEvStart(consumer), NodeIndex0, true);
+                Actors.insert(producer);
+                Actors.insert(consumer);
             } else {
-                auto producer = runtime.Register(new TWorkerActor(service1, TEvTestPrivate::ERole::Producer, channelId, producerSettings), nodeIndex1);
-                auto consumer = runtime.Register(new TWorkerActor(service0, TEvTestPrivate::ERole::Consumer, channelId, consumerSettings), nodeIndex0);
-                runtime.Send(consumer, control0, new TEvTestPrivate::TEvStart(producer), nodeIndex0, true);
-                runtime.Send(producer, control1, new TEvTestPrivate::TEvStart(consumer), nodeIndex1, true);
-                actors.insert(producer);
-                actors.insert(consumer);
+                auto producer = Runtime->Register(new TProducerActor(Service1, channelId, ProducerSettings), NodeIndex1);
+                auto consumer = Runtime->Register(new TConsumerActor(Service0, channelId, ConsumerSettings), NodeIndex0);
+                Runtime->Send(consumer, Control0, new TEvTestPrivate::TEvStart(producer), NodeIndex0, true);
+                Runtime->Send(producer, Control1, new TEvTestPrivate::TEvStart(consumer), NodeIndex1, true);
+                Actors.insert(producer);
+                Actors.insert(consumer);
             }
         }
+    }
 
-        int finishCount[2][2] = {{0, 0}, {0, 0}};
-        int errorCount = 0;
+    virtual void Wait() {
         try {
-            for (auto i = 0; i < count; i++) {
-                auto msg0 = runtime.GrabEdgeEvent<TEvTestPrivate::TEvFinished>(control0, TDuration::Seconds(10));
-                actors.erase(msg0->Sender);
-                finishCount[nodeIndex0][msg0->Get()->Role]++;
-                errorCount += msg0->Get()->Error;
-                auto msg1 = runtime.GrabEdgeEvent<TEvTestPrivate::TEvFinished>(control1, TDuration::Seconds(10));
-                actors.erase(msg1->Sender);
-                finishCount[nodeIndex1][msg1->Get()->Role]++;
-                errorCount += msg1->Get()->Error;
+            for (auto i = 0; i < Count; i++) {
+                auto msg0 = Runtime->GrabEdgeEvent<TEvTestPrivate::TEvFinished>(Control0, TDuration::Seconds(10));
+                Actors.erase(msg0->Sender);
+                FinishCount[NodeIndex0][msg0->Get()->Role]++;
+                ErrorCount += msg0->Get()->Error;
+                auto msg1 = Runtime->GrabEdgeEvent<TEvTestPrivate::TEvFinished>(Control1, TDuration::Seconds(10));
+                Actors.erase(msg1->Sender);
+                FinishCount[NodeIndex1][msg1->Get()->Role]++;
+                ErrorCount += msg1->Get()->Error;
             }
         } catch (NActors::TEmptyEventQueueException&) {
-            if (!actors.empty()) {
+            if (!Actors.empty()) {
                 TStringBuilder builder;
                 builder << "NOT FINISHED ACTORS ";
-                for (auto actorId : actors) {
+                for (auto actorId : Actors) {
                     builder << ' ' << actorId;
                 }
                 UNIT_ASSERT_C(false, builder);
             }
         }
+    }
 
-        if (local) {
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[0][TEvTestPrivate::ERole::Producer], count);
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[0][TEvTestPrivate::ERole::Consumer], count);
+    virtual void Check() {
+        if (Local) {
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[0][TEvTestPrivate::ERole::Producer], Count);
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[0][TEvTestPrivate::ERole::Consumer], Count);
         } else {
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[0][TEvTestPrivate::ERole::Producer], (count + 1) / 2);
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[0][TEvTestPrivate::ERole::Consumer], count / 2);
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[1][TEvTestPrivate::ERole::Producer], count / 2);
-            UNIT_ASSERT_VALUES_EQUAL(finishCount[1][TEvTestPrivate::ERole::Consumer], (count + 1) / 2);
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[0][TEvTestPrivate::ERole::Producer], (Count + 1) / 2);
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[0][TEvTestPrivate::ERole::Consumer], Count / 2);
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[1][TEvTestPrivate::ERole::Producer], Count / 2);
+            UNIT_ASSERT_VALUES_EQUAL(FinishCount[1][TEvTestPrivate::ERole::Consumer], (Count + 1) / 2);
         }
-        UNIT_ASSERT_VALUES_EQUAL(errorCount, 0);
+        UNIT_ASSERT_VALUES_EQUAL(ErrorCount, 0);
+    }
+
+    virtual void Run() {
+        Prepare();
+        Init();
+        Start();
+        Wait();
+        Check();
+    }
+
+    int Count = 1;
+    bool Local = true;
+    ui32 NodeIndex0 = 0;
+    ui32 NodeIndex1 = 1;
+    TKikimrSettings settings;
+    std::unique_ptr<TKikimrRunner> Runner;
+    NActors::TTestActorRuntime* Runtime;
+    std::shared_ptr<TDqChannelService> Service0;
+    std::shared_ptr<TDqChannelService> Service1;
+    NActors::TActorId Control0;
+    NActors::TActorId Control1;
+    TWorkerSettings ProducerSettings;
+    TWorkerSettings ConsumerSettings;
+    THashSet<NActors::TActorId> Actors;
+    int ErrorCount = 0;
+    int FinishCount[2][2] = {{0, 0}, {0, 0}};
+};
+
+struct TReconTest : public TLoadTest {
+
+    void Prepare() override {
+        TLoadTest::Prepare();
+        settings.AppConfig.MutableTableServiceConfig()->MutableDqChannelConfig()->SetCleanupPeriodMs(20);
+        settings.AppConfig.MutableTableServiceConfig()->MutableDqChannelConfig()->SetIdlePingPeriodMs(10);
+    }
+
+    void Start() override {
+        for (auto i = 0; i < Count; i ++) {
+            auto channelId = i + 1;
+            if ((i & 1) == 0) {
+                auto producerSettings = ProducerSettings;
+                producerSettings.PauseMessageIndex = (channelId + producerSettings.MessageCount / 2) % producerSettings.MessageCount;
+                producerSettings.PauseDelayMs = 50;
+                auto producer = Runtime->Register(new TProducerActor(Service0, channelId, producerSettings), NodeIndex0);
+                auto consumer = Runtime->Register(new TConsumerActor(Service1, channelId, ConsumerSettings), NodeIndex1);
+                Runtime->Send(consumer, Control1, new TEvTestPrivate::TEvStart(producer), NodeIndex1, true);
+                Runtime->Send(producer, Control0, new TEvTestPrivate::TEvStart(consumer), NodeIndex0, true);
+                Actors.insert(producer);
+                Actors.insert(consumer);
+            } else {
+                auto producerSettings = ProducerSettings;
+                producerSettings.PauseMessageIndex = channelId;
+                producerSettings.PauseDelayMs = 50;
+                auto producer = Runtime->Register(new TProducerActor(Service1, channelId, producerSettings), NodeIndex1);
+                auto consumer = Runtime->Register(new TConsumerActor(Service0, channelId, ConsumerSettings), NodeIndex0);
+                Runtime->Send(consumer, Control0, new TEvTestPrivate::TEvStart(producer), NodeIndex0, true);
+                Runtime->Send(producer, Control1, new TEvTestPrivate::TEvStart(consumer), NodeIndex1, true);
+                Actors.insert(producer);
+                Actors.insert(consumer);
+            }
+        }
+    }
+
+};
+
+Y_UNIT_TEST_SUITE(Channels20) {
+
+    void LoadTest(int count, bool local, const TWorkerSettings& producerSettings, const TWorkerSettings& consumerSettings, const TFailureSettings& = TFailureSettings{}) {
+        TLoadTest test;
+
+        test.Count = count;
+        test.Local = local;
+        test.ProducerSettings = producerSettings;
+        test.ConsumerSettings = consumerSettings;
+
+        test.Run();
     }
 
     void LoadTest(int count, bool local, const TWorkerSettings& settings = TWorkerSettings{}, const TFailureSettings& failureSettings = TFailureSettings{}) {
@@ -326,4 +410,15 @@ Y_UNIT_TEST_SUITE(Channels20) {
     Y_UNIT_TEST(MissedData) {
         LoadTest(100, false, TWorkerSettings{ .MessageCount = 100 }, TWorkerSettings{ .MessageCount = 100 }, TFailureSettings{ .Data = 10 });
     }
+
+    Y_UNIT_TEST(Reconciliation) {
+        TReconTest test;
+
+        test.Count = 1;
+        test.Local = false;
+        test.ProducerSettings.MessageCount = 100;
+        test.ConsumerSettings.MessageCount = 100;
+
+        test.Run();
+    }
 }
diff --git a/ydb/library/yql/dq/runtime/dq_channel_service.cpp b/ydb/library/yql/dq/runtime/dq_channel_service.cpp
index 4bb02dcb1fe..6446cd1d05a 100644
--- a/ydb/library/yql/dq/runtime/dq_channel_service.cpp
+++ b/ydb/library/yql/dq/runtime/dq_channel_service.cpp
@@ -1042,6 +1042,7 @@ void TNodeState::PushDataChunk(TDataChunk&& data, std::shared_ptr<TOutputDescrip
                 item->Leading = descriptor->Leading.exchange(false);
                 Queue.push_back(item);
                 SendMessage(item);
+                SendCount++;
                 InflightBytes += bytes;
                 *OutputBufferInflightBytes += bytes;
                 (*OutputBufferInflightMessages)++;
@@ -1080,7 +1081,7 @@ void TNodeState::PushDataChunk(TDataChunk&& data, std::shared_ptr<TOutputDescrip
 }
 
 void TNodeState::SendMessage(std::shared_ptr<TOutputItem> item) {
-    Y_ENSURE(PeerActorId);
+    Y_ENSURE(InputNodeActorId);
     auto ev = MakeHolder<TEvDqCompute::TEvChannelDataV2>();
 
     ev->Record.SetGenMajor(GenMajor);
@@ -1132,20 +1133,20 @@ void TNodeState::SendMessage(std::shared_ptr<TOutputItem> item) {
             FailureDoubleSend.store(failCount - 1);
             auto ev2 = MakeHolder<TEvDqCompute::TEvChannelDataV2>();
             ev2->Record = ev->Record;
-            ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, ev2.Release(), flags, item->SeqNo));
+            ActorSystem->Send(new NActors::IEventHandle(InputNodeActorId, NodeActorId, ev2.Release(), flags, item->SeqNo));
         }
 #endif
-        LOG_T(LogPrefix << "SEND MSG, G=" << GenMajor << '.' << GenMinor << ", SeqNo=" << item->SeqNo
-            << ", ChannelSeqNo=" << item->ChannelSeqNo
-            << ", ChannelId=" << item->Descriptor->Info.ChannelId << ", Leading=" << item->Leading << ", Bytes=" << item->Data.Bytes);
-        ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, ev.Release(), flags, item->SeqNo));
+        LOG_T(LogPrefix << "SEND DATA, G=" << GenMajor << '.' << GenMinor << ", SeqNo=" << item->SeqNo
+            << ", ChannelSeqNo=" << item->ChannelSeqNo << ", ChannelId=" << item->Descriptor->Info.ChannelId
+            << ", Leading=" << item->Leading << ", Finished=" << item->Data.Finished << ", Bytes=" << item->Data.Bytes);
+        ActorSystem->Send(new NActors::IEventHandle(InputNodeActorId, NodeActorId, ev.Release(), flags, item->SeqNo));
 #if !defined(NDEBUG)
     }
 #endif
     item->State.store(TOutputItem::EState::Sent);
 }
 
-void TNodeState::FailInputs(const NActors::TActorId& peerActorId, ui64 peerGenMajor) {
+void TNodeState::FailInputs(const NActors::TActorId& outputNodeActorId, ui64 outputNodeGenMajor) {
     if (InputDescriptors.empty()) {
         return;
     }
@@ -1153,11 +1154,11 @@ void TNodeState::FailInputs(const NActors::TActorId& peerActorId, ui64 peerGenMa
     std::vector<TChannelInfo> failedBuffers;
 
     for (auto& [info, descriptor] : InputDescriptors) {
-        if (!descriptor->IsFinished() && descriptor->PeerGenMajor) {
-            if (descriptor->PeerActorId != peerActorId || descriptor->PeerGenMajor != peerGenMajor) {
+        if (!descriptor->IsFinished() && descriptor->OutputNodeGenMajor) {
+            if (descriptor->OutputNodeActorId != outputNodeActorId || descriptor->OutputNodeGenMajor != outputNodeGenMajor) {
                 descriptor->AbortChannel(
-                    TStringBuilder() << "PeerActorId=" << descriptor->PeerActorId << ", PeerGenMajor=" << descriptor->PeerGenMajor
-                    << " DO NOT MATCH peerActorId=" << peerActorId << ", peerGenMajor=" << peerGenMajor
+                    TStringBuilder() << "OutputNodeActorId=" << descriptor->OutputNodeActorId << ", OutputNodeGenMajor=" << descriptor->OutputNodeGenMajor
+                    << " DO NOT MATCH outputNodeActorId=" << outputNodeActorId << ", outputNodeGenMajor=" << outputNodeGenMajor
                     << ", Session=" << LogPrefix << ", Log=" << GetReconciliationLog()
                 );
                 failedBuffers.push_back(info);
@@ -1207,14 +1208,14 @@ void TNodeState::SendAck(THolder<TEvDqCompute::TEvChannelAckV2>& evAck, ui64 coo
         flags |=  NActors::IEventHandle::FlagSubscribeOnSession;
     }
 
-    ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, evAck.Release(), flags, cookie));
+    ActorSystem->Send(new NActors::IEventHandle(OutputNodeActorId, NodeActorId, evAck.Release(), flags, cookie));
 }
 
 void TNodeState::SendAckWithError(ui64 cookie, const TString& message) {
     auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
 
-    evAck->Record.SetGenMajor(PeerGenMajor.load());
-    evAck->Record.SetGenMinor(PeerGenMinor.load());
+    evAck->Record.SetGenMajor(OutputNodeGenMajor.load());
+    evAck->Record.SetGenMinor(OutputNodeGenMinor.load());
     evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::ERROR);
     evAck->Record.SetSeqNo(ConfirmedSeqNo);
     evAck->Record.SetMessage(message);
@@ -1245,12 +1246,12 @@ void TNodeState::HandleChannelData(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
         return;
     }
 
-    if (descriptor->PeerGenMajor) {
-        if (descriptor->PeerActorId != PeerActorId || descriptor->PeerGenMajor != PeerGenMajor.load()) {
+    if (descriptor->OutputNodeGenMajor) {
+        if (descriptor->OutputNodeActorId != OutputNodeActorId || descriptor->OutputNodeGenMajor != OutputNodeGenMajor.load()) {
             descriptor->Terminate();
-            TString errorMessage = TStringBuilder() << "MISMATCH G=" << descriptor->PeerGenMajor
-                << ", Peer=" << descriptor->PeerGenMajor << " vs G=" << PeerGenMajor.load()
-                << ", Peer=" << PeerActorId
+            TString errorMessage = TStringBuilder() << "MISMATCH ID G=" << descriptor->OutputNodeGenMajor
+                << ", Peer=" << descriptor->OutputNodeGenMajor << " vs G=" << OutputNodeGenMajor.load()
+                << ", Peer=" << OutputNodeActorId
                 << ", ChannelId=" << info.ChannelId
                 << ", OA=" << info.OutputActorId << ", IA=" << info.InputActorId
                 << ", Bytes=" << record.GetBytes() << ", L=" << record.GetLeading()
@@ -1264,8 +1265,8 @@ void TNodeState::HandleChannelData(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
             return;
         }
     } else {
-        descriptor->PeerActorId = PeerActorId;
-        descriptor->PeerGenMajor = PeerGenMajor.load();
+        descriptor->OutputNodeActorId = OutputNodeActorId;
+        descriptor->OutputNodeGenMajor = OutputNodeGenMajor.load();
     }
 
     TDataChunk data(TChunkedBuffer(), record.GetRows(), record.GetTransportVersion(),
@@ -1322,8 +1323,8 @@ void TNodeState::HandleChannelData(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
 
     auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
 
-    evAck->Record.SetGenMajor(PeerGenMajor.load());
-    evAck->Record.SetGenMinor(PeerGenMinor.load());
+    evAck->Record.SetGenMajor(OutputNodeGenMajor.load());
+    evAck->Record.SetGenMinor(OutputNodeGenMinor.load());
     evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::OK);
     evAck->Record.SetSeqNo(ConfirmedSeqNo);
 
@@ -1346,25 +1347,18 @@ void TNodeState::HandleDisconnected(NActors::TEvInterconnect::TEvNodeDisconnecte
 
 void TNodeState::HandleUndelivered(NActors::TEvents::TEvUndelivered::TPtr& ev) {
 
-    if (ev->Get()->Reason == NActors::TEvents::TEvUndelivered::ReasonActorUnknown) {
-        std::lock_guard lock(Mutex);
-        if (Reconciliation.load() == 0) { // ignore errors in recovery
-            if (ev->Get()->SourceType == TEvDqCompute::TEvChannelUpdateV2::EventType) {
-                LOG_D(LogPrefix << "UNDELIVERED/UNKNOWN UPDATE, PeerActorId " << PeerActorId << ", Sender=" << ev->Sender);
-            } else {
-                LOG_W(LogPrefix << "UNDELIVERED/UNKNOWN, PeerActorId " << PeerActorId << ", Sender=" << ev->Sender);
-                PeerActorId = NActors::TActorId{};
-                StartReconciliation(true, 'U');
-            }
-        }
-        return;
-    }
-
     switch (ev->Get()->SourceType) {
         case TEvDqCompute::TEvChannelDataV2::EventType: {
-            LOG_W(LogPrefix << "UNDELIVERED/OTHER");
             std::lock_guard lock(Mutex);
-            StartReconciliation(false, 'O');
+            if (ev->Get()->Reason == NActors::TEvents::TEvUndelivered::ReasonActorUnknown) {
+                if (Reconciliation.load() == 0) { // ignore errors in recovery
+                    LOG_W(LogPrefix << "UNDELIVERED/UNKNOWN, InputNodeActorId " << InputNodeActorId << ", Sender=" << ev->Sender);
+                    StartReconciliation(true, 'U');
+                }
+            } else {
+                LOG_W(LogPrefix << "UNDELIVERED/OTHER");
+                StartReconciliation(false, 'O');
+            }
             break;
         }
         case TEvDqCompute::TEvChannelAckV2::EventType: {
@@ -1380,45 +1374,32 @@ void TNodeState::HandleUndelivered(NActors::TEvents::TEvUndelivered::TPtr& ev) {
     }
 }
 
-void TNodeState::ConnectSession(NActors::TActorId& sender, ui64 genMajor, ui64 genMinor, ui64 seqNo) {
-    std::lock_guard lock(Mutex);
-    if (!Connected) {
-        PeerActorId = sender;
-        PeerGenMajor.store(genMajor);
-        PeerGenMinor.store(genMinor);
-        ConfirmedSeqNo = seqNo;
-        Connected = true;
-        LOG_D(LogPrefix << "CONNECTED, PeerActorId=" << sender << ", PG=" << genMajor << '.' << genMinor);
+void TNodeState::ConnectSession(NActors::TActorId& sender, ui64 genMajor, ui64 genMinor) {
+    if (OutputNodeActorId == sender && OutputNodeGenMajor.load() == genMajor) {
+        LOG_D(LogPrefix << "RECONNECTED, OutputNodeActorId=" << sender << ", PG=" << genMajor << '.' << genMinor);
     } else {
-        if (PeerActorId == sender && PeerGenMajor.load() == genMajor) {
-            LOG_D(LogPrefix << "RECONNECTED, PeerActorId=" << sender << ", PG=" << genMajor << '.' << genMinor);
-        } else {
-            PeerActorId = sender;
-            PeerGenMajor.store(genMajor);
-            LOG_W(LogPrefix << "RECONNECTED, PeerActorId=" << sender << ", PG=" << genMajor << '.' << genMinor);
-        }
-        PeerGenMinor.store(genMinor);
-        ConfirmedSeqNo = seqNo;
-        FailInputs(PeerActorId, PeerGenMajor.load());
-    }
-    for (auto& [_, descriptor] : InputDescriptors) {
-        if (descriptor->EarlyFinished.load() || descriptor->PushStats.Bytes.load()) {
-            SendUpdateProgress(descriptor);
-        }
+        OutputNodeActorId = sender;
+        OutputNodeGenMajor.store(genMajor);
+        ConfirmedSeqNo = 0;
+        LOG_W(LogPrefix << "RECONNECTED, OutputNodeActorId=" << sender << ", PG=" << genMajor << '.' << genMinor);
     }
+    OutputNodeGenMinor.store(genMinor);
+    FailInputs(OutputNodeActorId, OutputNodeGenMajor.load());
 }
 
 void TNodeState::HandleDiscovery(TEvDqCompute::TEvChannelDiscoveryV2::TPtr& ev) {
 
     LastPeerActivity.store(TInstant::Now());
     auto& record = ev->Get()->Record;
-    ConnectSession(ev->Sender, record.GetGenMajor(), record.GetGenMinor(), record.GetSeqNo());
+
+    std::lock_guard lock(Mutex);
+    ConnectSession(ev->Sender, record.GetGenMajor(), record.GetGenMinor());
 
     auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
 
-    evAck->Record.SetGenMajor(PeerGenMajor.load());
-    evAck->Record.SetGenMinor(PeerGenMinor.load());
-    evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::OK);
+    evAck->Record.SetGenMajor(OutputNodeGenMajor.load());
+    evAck->Record.SetGenMinor(OutputNodeGenMinor.load());
+    evAck->Record.SetStatus(record.GetSeqNo() <= ConfirmedSeqNo ? NYql::NDqProto::TEvChannelAckV2::OK : NYql::NDqProto::TEvChannelAckV2::RESEND);
     evAck->Record.SetSeqNo(ConfirmedSeqNo);
 
     ui32 flags = NActors::IEventHandle::FlagTrackDelivery;
@@ -1426,67 +1407,46 @@ void TNodeState::HandleDiscovery(TEvDqCompute::TEvChannelDiscoveryV2::TPtr& ev)
         flags |=  NActors::IEventHandle::FlagSubscribeOnSession;
     }
 
-    ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, evAck.Release(), flags, ev->Cookie));
+    ActorSystem->Send(new NActors::IEventHandle(OutputNodeActorId, NodeActorId, evAck.Release(), flags, ev->Cookie));
+
+    for (auto& [_, descriptor] : InputDescriptors) {
+        if (descriptor->EarlyFinished.load() || descriptor->PushStats.Bytes.load()) {
+            SendUpdateProgress(descriptor);
+        }
+    }
 }
 
 void TNodeState::HandleData(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
 
-    LastPeerActivity.store(TInstant::Now());
     auto& record = ev->Get()->Record;
-    LOG_T(LogPrefix << "RECV MSG, PG=" << PeerGenMajor.load() << '.' << PeerGenMinor.load() << ", SeqNo=" << record.GetSeqNo()
-        << ", ChannelSeqNo=" << record.GetChannelSeqNo()
-        << ", ChannelId=" << record.GetChannelId() << ", Bytes=" << record.GetBytes());
-
-    auto prevPeerGenMajor = PeerGenMajor.exchange(record.GetGenMajor());
-    if (PeerActorId != ev->Sender || (prevPeerGenMajor && prevPeerGenMajor != record.GetGenMajor())) {
-        auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
-
-        auto reconciliationLog = GetReconciliationLog();
-        evAck->Record.SetGenMajor(record.GetGenMajor());
-        evAck->Record.SetGenMinor(record.GetGenMinor());
-        evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::FAIL);
-        evAck->Record.SetSeqNo(ConfirmedSeqNo);
-        evAck->Record.SetMessage(reconciliationLog);
-        LOG_E(LogPrefix << "FAIL, PeerActorId=" << PeerActorId << ", ev->Sender" << ev->Sender
-            << ", PeerGenMajor=" << prevPeerGenMajor << ", record.GetGenMajor()=" << record.GetGenMajor()
-            << ", Log=" << reconciliationLog);
-
-        ui32 flags = NActors::IEventHandle::FlagTrackDelivery;
-        if (!Subscribed.exchange(true)) {
-            flags |=  NActors::IEventHandle::FlagSubscribeOnSession;
-        }
+    auto genMajor = record.GetGenMajor();
+    auto genMinor = record.GetGenMinor();
+    auto seqNo = record.GetSeqNo();
 
-        ActorSystem->Send(new NActors::IEventHandle(ev->Sender, NodeActorId, evAck.Release(), flags, ev->Cookie));
-        return;
+    if (OutputNodeActorId != ev->Sender || OutputNodeGenMajor.load() != genMajor || OutputNodeGenMinor.load() != genMinor) {
+        LOG_D(LogPrefix << "OBSOLETE DATA, OutputNodeActorId=" << OutputNodeActorId << ", G=" << OutputNodeGenMajor.load() << '.' << OutputNodeGenMinor.load()
+            << " vs Sender=" << ev->Sender << ", G=" << genMajor << '.' << genMinor
+            << ", SeqNo=" << seqNo);
+    } else {
+        LOG_T(LogPrefix << "RECV DATA, G=" << genMajor << '.' << genMinor << ", SeqNo=" << seqNo
+            << ", ChannelSeqNo=" << record.GetChannelSeqNo() << ", ChannelId=" << record.GetChannelId()
+            << ", Leading=" << record.GetLeading() << ", Finished=" << record.GetFinished()
+            << ", Bytes=" << record.GetBytes());
     }
 
-    PeerGenMinor.store(std::max<ui64>(record.GetGenMinor(), PeerGenMinor.load()));
-
-    auto seqNo = record.GetSeqNo();
-
     if (seqNo <= ConfirmedSeqNo) {
         LOG_W(LogPrefix << "DATA/IGNORED, SeqNo=" << seqNo << ", ConfirmedSeqNo=" << ConfirmedSeqNo);
         return;
     }
 
-    switch (seqNo - ConfirmedSeqNo) {
-        case 1: {
-            break;
-        }
-        case 2: {
-            // allow 1 out of order message
-            LOG_W(LogPrefix << "DATA/OOO, SeqNo=" << seqNo << ", ConfirmedSeqNo=" << ConfirmedSeqNo);
-            OutOfOrderMessage = ev.Release();
-            return;
-        }
-        default: {
-            if (!ResendAsked.exchange(true)) {
-                LOG_W(LogPrefix << "DATA/RESEND, SeqNo=" << seqNo << ", ConfirmedSeqNo=" << ConfirmedSeqNo);
-            }
+    if (seqNo - ConfirmedSeqNo > 1) {
+        if (!ResendAsked.exchange(true)) {
+            LOG_W(LogPrefix << "DATA/RESEND, SeqNo=" << seqNo << ", ConfirmedSeqNo=" << ConfirmedSeqNo);
+
             auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
 
-            evAck->Record.SetGenMajor(PeerGenMajor.load());
-            evAck->Record.SetGenMinor(PeerGenMinor.load());
+            evAck->Record.SetGenMajor(OutputNodeGenMajor.load());
+            evAck->Record.SetGenMinor(OutputNodeGenMinor.load());
             evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::RESEND);
             evAck->Record.SetSeqNo(ConfirmedSeqNo + 1);
 
@@ -1495,27 +1455,15 @@ void TNodeState::HandleData(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
                 flags |=  NActors::IEventHandle::FlagSubscribeOnSession;
             }
 
-            ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, evAck.Release(), flags, ev->Cookie));
-            return;
+            ActorSystem->Send(new NActors::IEventHandle(OutputNodeActorId, NodeActorId, evAck.Release(), flags, ev->Cookie));
         }
+        return;
     }
 
     // happy path
-
     ResendAsked.store(false);
     ConfirmedSeqNo++;
     HandleChannelData(ev);
-
-    if (OutOfOrderMessage) {
-        auto& record = OutOfOrderMessage->Get()->Record;
-
-        if (record.GetSeqNo() == ConfirmedSeqNo + 1) {
-            ConfirmedSeqNo++;
-            HandleChannelData(OutOfOrderMessage);
-        }
-
-        OutOfOrderMessage.Reset();
-    }
 }
 
 void TNodeState::SendFromWaiters(ui64 deltaBytes) {
@@ -1540,6 +1488,7 @@ void TNodeState::SendFromWaiters(ui64 deltaBytes) {
 
                 (*OutputBufferWaiterCount)--;
 /*
+now may need to send very last msg from terminated descriptor
                 if (WaitersQueue.top()->IsTerminatedOrAborted()) {
 
                     auto waitQueueBytes = WaitersQueue.top()->WaitQueueBytes.load();
@@ -1599,6 +1548,7 @@ void TNodeState::SendFromWaiters(ui64 deltaBytes) {
                 item->Leading = waiter->Leading.exchange(false);
                 Queue.push_back(item);
                 SendMessage(item);
+                SendCount++;
                 inflightBytes += bytes;
                 InflightBytes += bytes;
                 *OutputBufferInflightBytes += bytes;
@@ -1645,7 +1595,7 @@ void TNodeState::HandleAck(TEvDqCompute::TEvChannelAckV2::TPtr& ev) {
         auto genMajor = record.GetGenMajor();
         auto genMinor = record.GetGenMinor();
         if (GenMajor != genMajor || GenMinor != genMinor) {
-            LOG_W(LogPrefix << "ACK/IGNORED, G=" << GenMajor << '.' << GenMinor << ", ack.G=" << genMajor << '.' << genMinor);
+            LOG_W(LogPrefix << "ACK/IGNORED, G=" << GenMajor << '.' << GenMinor << ", ack.G=" << genMajor << '.' << genMinor << ", Sender=" << ev->Sender);
             return;
         }
 
@@ -1658,9 +1608,9 @@ void TNodeState::HandleAck(TEvDqCompute::TEvChannelAckV2::TPtr& ev) {
             return;
         }
 
-        if (PeerActorId == NActors::TActorId{}) {
-            PeerActorId = ev->Sender;
-            LOG_D(LogPrefix << "PEER/ACK, PeerActorId=" << ev->Sender);
+        if (InputNodeActorId == NActors::TActorId{}) {
+            InputNodeActorId = ev->Sender;
+            LOG_D(LogPrefix << "PEER/ACK, InputNodeActorId=" << ev->Sender);
         }
 
         if (seqNo > SeqNo) {
@@ -1701,11 +1651,8 @@ void TNodeState::HandleAck(TEvDqCompute::TEvChannelAckV2::TPtr& ev) {
                 }
             } else {
                 if (status == NYql::NDqProto::TEvChannelAckV2::RESEND) {
-                    // if we're reconcilating, ignore next RESENDs
-                    if (record.GetGenMinor() == GenMinor) {
-                        LOG_W(LogPrefix << "SEQ/RESEND, SeqNo=" << seqNo);
-                        StartReconciliation(false, 'R');
-                    }
+                    LOG_W(LogPrefix << "SEQ/RESEND, SeqNo=" << seqNo);
+                    StartReconciliation(false, 'R');
                     return;
                 }
 
@@ -1735,10 +1682,12 @@ void TNodeState::HandleAck(TEvDqCompute::TEvChannelAckV2::TPtr& ev) {
 
         if (Reconciliation.exchange(0) > 0) {
             ReconciliationCount = 0;
+            ReconSent.store(TInstant::Zero());
             LOG_I(LogPrefix << "RECONCILED, Q=" << (Queue.empty() ? "E" : ToString(Queue.front()->SeqNo)) << ':' << SeqNo << ", WQ=" << WaitersQueueSize.load() << ", InflightBytes=" << InflightBytes.load() << '-' << deltaBytes);
             if (!Queue.empty()) {
                 for (auto item : Queue) {
                     SendMessage(item);
+                    ResendCount++;
                     item->Descriptor->CheckGenMajor(GenMajor, TStringBuilder() << "Abort by Repeat from SeqNo=" << Queue.front()->SeqNo << ", item->SeqNo=" << item->SeqNo);
                 }
             }
@@ -1817,8 +1766,8 @@ void TNodeState::SendUpdateProgress(std::shared_ptr<TInputDescriptor>& descripto
 
     auto evUpdate = MakeHolder<TEvDqCompute::TEvChannelUpdateV2>();
 
-    evUpdate->Record.SetGenMajor(PeerGenMajor.load());
-    evUpdate->Record.SetGenMinor(PeerGenMinor.load());
+    evUpdate->Record.SetGenMajor(OutputNodeGenMajor.load());
+    evUpdate->Record.SetGenMinor(OutputNodeGenMinor.load());
     // evUpdate->Record.SetSeqNo(ConfirmedSeqNo);
 
     NActors::ActorIdToProto(descriptor->Info.OutputActorId, evUpdate->Record.MutableSrcActorId());
@@ -1839,7 +1788,7 @@ void TNodeState::SendUpdateProgress(std::shared_ptr<TInputDescriptor>& descripto
         << ", EarlyFinished=" << descriptor->EarlyFinished.load() << ", PopBytes=" << descriptor->PopStats.Bytes.load()
         << ", Finishing=" << descriptor->Finishing.load());
 
-    ActorSystem->Send(new NActors::IEventHandle(PeerActorId, NodeActorId, evUpdate.Release(), flags));
+    ActorSystem->Send(new NActors::IEventHandle(OutputNodeActorId, NodeActorId, evUpdate.Release(), flags));
 }
 
 std::shared_ptr<TOutputDescriptor> TNodeState::GetOrCreateOutputDescriptor(const TChannelFullInfo& info, IMemoryQuotaManager::TPtr quotaManager, bool bound, bool leading) {
@@ -2035,10 +1984,12 @@ void TNodeState::HandleReconciliation(TEvPrivate::TEvReconciliation::TPtr& ev) {
 
 void TNodeState::StartReconciliation(bool major, char logSymbol) {
     if (Reconciliation.load() == 0 || (major && (GenMinor > 1))) {
+        ReconCount++;
         if (major) {
             GenMajor++;
             GenMinor = 1;
             SeqNo = 0;
+            InputNodeActorId = NActors::TActorId{};
         } else {
             GenMinor++;
         }
@@ -2087,70 +2038,70 @@ void TNodeState::DoReconciliation(char logSymbol) {
             << ", Q=" << (Queue.empty() ? "E" : ToString(Queue.front()->SeqNo)) << ':' << SeqNo
             << ", WQ=" << WaitersQueueSize.load() << ", Log=" << reconciliationLog);
     } else {
-        LOG_D(LogPrefix << "RECONCILIATION, G" << GenMajor << '.' << GenMinor
+        LOG_D(LogPrefix << "RECONCILIATION, G=" << GenMajor << '.' << GenMinor
             << ", Q=" << (Queue.empty() ? "E" : ToString(Queue.front()->SeqNo)) << ':' << SeqNo
             << ", WQ=" << WaitersQueueSize.load() << ", Log=" << reconciliationLog);
     }
 
     ui32 delta = 0;
 
-    std::deque<std::shared_ptr<TOutputItem>> RebuiltQueue;
-
-    auto seqNo = SeqNo;
+    if (GenMinor == 1) { // => major reconciliation
+        std::deque<std::shared_ptr<TOutputItem>> RebuiltQueue;
+        while (!Queue.empty()) {
 
-    while (!Queue.empty()) {
+            auto& item = Queue.front();
 
-        auto& item = Queue.front();
+            if (item->Leading) {
+                auto prevGenMajor = item->Descriptor->GenMajor.exchange(GenMajor);
+                if (prevGenMajor != GenMajor) {
+                    LOG_W(LogPrefix << "CHANGE OD G=" << prevGenMajor << " to " << GenMajor << " by RECONCILIATION"
+                        << " Channel: " << item->Descriptor->Info.ChannelId
+                        << ", SrcStageId: " << item->Descriptor->Info.SrcStageId
+                        << ", DstStageId: " << item->Descriptor->Info.DstStageId
+                        << ", Log=" << reconciliationLog);
+                }
+            }
 
-        if (item->Leading) {
-            auto prevGenMajor = item->Descriptor->GenMajor.exchange(GenMajor);
-            if (prevGenMajor != GenMajor) {
-                LOG_W(LogPrefix << "CHANGE OG G=" << prevGenMajor << " to " << GenMajor << " by RECONCILIATION"
-                    << " Channel: " << item->Descriptor->Info.ChannelId
-                    << ", SrcStageId: " << item->Descriptor->Info.SrcStageId
-                    << ", DstStageId: " << item->Descriptor->Info.DstStageId
-                    << ", Log=" << reconciliationLog);
+            if (item->Descriptor->CheckGenMajor(GenMajor, TStringBuilder() << "Abort by Reconciliation, Log=" << reconciliationLog)) {
+                item->SeqNo = ++SeqNo;
+                RebuiltQueue.push_back(std::move(item));
+            } else {
+                delta += item->Data.Bytes;
             }
-        }
 
-        if (item->Descriptor->CheckGenMajor(GenMajor, TStringBuilder() << "Abort by Reconciliation, Log=" << reconciliationLog)) {
-            item->SeqNo = ++SeqNo;
-            RebuiltQueue.push_back(std::move(item));
-        } else {
-            delta += item->Data.Bytes;
+            Queue.pop_front();
         }
-
-        Queue.pop_front();
+        Queue.swap(RebuiltQueue);
     }
-    Queue.swap(RebuiltQueue);
 
     InflightBytes -= delta;
 
-    SendDiscovery(MakeChannelServiceActorID(NodeId), seqNo);
+    SendDiscovery();
+    ReconSent.store(TInstant::Now());
     ActorSystem->Schedule(reconciliationTimeout,
         new NActors::IEventHandle(NodeActorId, NodeActorId, new TEvPrivate::TEvReconciliation(GenMajor, GenMinor, ReconciliationCount)));
 }
 
-void TNodeState::SendDiscovery(NActors::TActorId actorId, ui64 seqNo) {
+void TNodeState::SendDiscovery() {
     auto evDiscovery = MakeHolder<TEvDqCompute::TEvChannelDiscoveryV2>();
 
     evDiscovery->Record.SetGenMajor(GenMajor);
     evDiscovery->Record.SetGenMinor(GenMinor);
-    evDiscovery->Record.SetSeqNo(seqNo);
+    evDiscovery->Record.SetSeqNo(SeqNo);
 
     ui32 flags = NActors::IEventHandle::FlagTrackDelivery;
     if (!Subscribed.exchange(true)) {
         flags |=  NActors::IEventHandle::FlagSubscribeOnSession;
     }
 
-    ActorSystem->Send(new NActors::IEventHandle(actorId, NodeActorId, evDiscovery.Release(), flags));
+    ActorSystem->Send(new NActors::IEventHandle(MakeChannelServiceActorID(NodeId), NodeActorId, evDiscovery.Release(), flags));
 }
 
 TString TNodeState::GetDebugInfo() {
     std::lock_guard lock(Mutex);
     TStringBuilder builder;
 
-    builder << "TNodeState, NodeId=" << NodeActorId.NodeId() << ", Peer NodeId=" << PeerActorId.NodeId()
+    builder << "TNodeState, NodeId=" << NodeActorId.NodeId() << ", Peer NodeId=" << NodeId
         << ", SeqNo=" << SeqNo << ", ConfirmedSeqNo=" << ConfirmedSeqNo << ", InflightBytes=" << InflightBytes.load()
         << ", Reconciliation=" << Reconciliation.load()
         << Endl;
@@ -2175,7 +2126,7 @@ void TDebugNodeState::HandleNullMode(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
 
     auto& record = ev->Get()->Record;
 
-    PeerGenMinor.store(std::max<ui64>(record.GetGenMinor(), PeerGenMinor.load()));
+    OutputNodeGenMinor.store(record.GetGenMinor());
 
     auto seqNo = record.GetSeqNo();
 
@@ -2196,8 +2147,8 @@ void TDebugNodeState::HandleNullMode(TEvDqCompute::TEvChannelDataV2::TPtr& ev) {
 
     auto evAck = MakeHolder<TEvDqCompute::TEvChannelAckV2>();
 
-    evAck->Record.SetGenMajor(PeerGenMajor.load());
-    evAck->Record.SetGenMinor(PeerGenMinor.load());
+    evAck->Record.SetGenMajor(OutputNodeGenMajor.load());
+    evAck->Record.SetGenMinor(OutputNodeGenMinor.load());
     evAck->Record.SetStatus(NYql::NDqProto::TEvChannelAckV2::OK);
     evAck->Record.SetSeqNo(ConfirmedSeqNo);
 
@@ -2396,6 +2347,7 @@ IDqInputChannel::TPtr TDqChannelService::GetInputChannel(const TDqChannelSetting
 }
 
 void TDqChannelService::NotifyCleanup() {
+    LOG_T("CLEANUP");
     ActorSystem->Schedule(Limits.CleanupPeriod, new NActors::IEventHandle(ServiceActorId, ServiceActorId, new TEvPrivate::TEvCleanup()));
     std::lock_guard lock(Mutex);
     for (auto& [_, nodeState] : NodeStates) {
@@ -2611,18 +2563,22 @@ void TChannelServiceActor::Handle(NActors::NMon::TEvHttpInfo::TPtr& ev) {
                         TABLEH_ATTRS({{"title", "FailureReconciliation"}}) {str << "FR";}
 #endif
                         TABLEH_ATTRS({{"title", "FailureDestroy"}}) {str << "F~";}
-                        TABLEH_ATTRS({{"title", "GenMajor"}}) {str << "GM";}
-                        TABLEH_ATTRS({{"title", "GenMinor"}}) {str << "gm";}
+                        TABLEH_ATTRS({{"title", "Gen"}}) {str << "G";}
+                        TABLEH_ATTRS({{"title", "Queue.front()->SeqNo"}}) {str << "f/SeqNo";}
                         TABLEH() {str << "SeqNo";}
-                        TABLEH_ATTRS({{"title", "Queue.front()->SeqNo"}}) {str << "front()";}
                         TABLEH_ATTRS({{"title", "InflightBytes"}}) {str << "InflightB";}
+                        TABLEH_ATTRS({{"title", "SendCount"}}) {str << "Send";}
+                        TABLEH_ATTRS({{"title", "ResendCount"}}) {str << "Resend";}
+                        TABLEH_ATTRS({{"title", "ReconCount"}}) {str << "Recon";}
+                        TABLEH() {str << "ReconSent";}
                         TABLEH_ATTRS({{"title", "WaitersQueueSize"}}) {str << "W/Queue";}
                         TABLEH_ATTRS({{"title", "WaitersMessages"}}) {str << "W/Msg";}
-                        TABLEH_ATTRS({{"title", "PeerGenMajor"}}) {str << "PM";}
-                        TABLEH_ATTRS({{"title", "PeerGenMinor"}}) {str << "pm";}
+                        TABLEH_ATTRS({{"title", "OutputNodeGen"}}) {str << "OG";}
                         TABLEH_ATTRS({{"title", "ConfirmedSeqNo"}}) {str << "C/SeqNo";}
-                        TABLEH() {str << "PeerActorId";}
+                        TABLEH() {str << "OutputNodeActorId";}
+                        TABLEH() {str << "InputNodeActorId";}
                         TABLEH() {str << "NodeActorId";}
+                        TABLEH() {str << "ReconciliationLog";}
                     }
                 }
                 TABLEBODY() {
@@ -2656,22 +2612,26 @@ void TChannelServiceActor::Handle(NActors::NMon::TEvHttpInfo::TPtr& ev) {
                                     str << "X";
                                 }
                             }
-                            TABLED() {str << state->GenMajor;}
-                            TABLED() {str << state->GenMinor;}
-                            TABLED() {str << state->SeqNo;}
+                            TABLED() {str << state->GenMajor << '.' << state->GenMinor;}
                             TABLED() {
                                 if (!state->Queue.empty()) {
                                     str << state->Queue.front()->SeqNo;
                                 }
                             }
+                            TABLED() {str << state->SeqNo;}
                             TABLED() {str << state->InflightBytes.load();}
+                            TABLED() {str << state->SendCount.load();}
+                            TABLED() {str << state->ResendCount.load();}
+                            TABLED() {str << state->ReconCount.load();}
+                            TABLED() {str << state->ReconSent.load();}
                             TABLED() {str << state->WaitersQueue.size();}
                             TABLED() {str << state->WaiterMessages.load();}
-                            TABLED() {str << state->PeerGenMajor.load();}
-                            TABLED() {str << state->PeerGenMinor.load();}
+                            TABLED() {str << state->OutputNodeGenMajor.load() << '.' << state->OutputNodeGenMinor.load();}
                             TABLED() {str << state->ConfirmedSeqNo;}
-                            TABLED() {str << state->PeerActorId;}
+                            TABLED() {str << state->OutputNodeActorId;}
+                            TABLED() {str << state->InputNodeActorId;}
                             TABLED() {str << state->NodeActorId;}
+                            TABLED() {str << state->GetReconciliationLog();}
                         }
                     }
                 }
diff --git a/ydb/library/yql/dq/runtime/dq_channel_service_impl.h b/ydb/library/yql/dq/runtime/dq_channel_service_impl.h
index 032f9412e1b..b1e43f6194d 100644
--- a/ydb/library/yql/dq/runtime/dq_channel_service_impl.h
+++ b/ydb/library/yql/dq/runtime/dq_channel_service_impl.h
@@ -439,8 +439,8 @@ public:
 
     TChannelFullInfo Info;
     NActors::TActorSystem* ActorSystem;
-    ui64 PeerGenMajor = 0;
-    NActors::TActorId PeerActorId;
+    ui64 OutputNodeGenMajor = 0;
+    NActors::TActorId OutputNodeActorId;
     bool IsBound = false;
     TDqThreadSafeStats PushStats;
     TDqThreadSafeStats PopStats;
@@ -586,13 +586,13 @@ public:
     void TerminateOutputDescriptor(const std::shared_ptr<TOutputDescriptor>& descriptor);
     void TerminateInputDescriptor(const std::shared_ptr<TInputDescriptor>& descriptor);
     void HandleCleanup();
-    void FailInputs(const NActors::TActorId& peerActorId, ui64 peerGenMajor);
+    void FailInputs(const NActors::TActorId& outputNodeActorId, ui64 outputNodeGenMajor);
     void FailOutputs(const NActors::TActorId& peerActorId, ui64 peerGenMajor);
     void SendAck(THolder<TEvDqCompute::TEvChannelAckV2>& evAck, ui64 cookie);
     void SendAckWithError(ui64 cookie, const TString& message);
     void HandleChannelData(TEvDqCompute::TEvChannelDataV2::TPtr& ev);
     void SendFromWaiters(ui64 deltaBytes);
-    void ConnectSession(NActors::TActorId& sender, ui64 genMajor, ui64 genMinor, ui64 seqNo);
+    void ConnectSession(NActors::TActorId& sender, ui64 genMajor, ui64 genMinor);
     virtual TString GetDebugInfo();
     void UpdateProgress(std::shared_ptr<TInputDescriptor>& descriptor);
     void SendUpdateProgress(std::shared_ptr<TInputDescriptor>& descriptor);
@@ -602,7 +602,7 @@ public:
     void DoReconciliation(char logSymbol);
     void AddReconciliationLog(char logSymbol);
     TString GetReconciliationLog();
-    void SendDiscovery(NActors::TActorId actorId, ui64 seqNo);
+    void SendDiscovery();
 
     NActors::TActorId NodeActorId;
     TString LogPrefix;
@@ -615,7 +615,6 @@ public:
     mutable std::unordered_map<TChannelInfo, std::shared_ptr<TInputDescriptor>> InputDescriptors;
     mutable std::queue<std::pair<TChannelInfo, TInstant>> UnboundInputs;
     mutable std::queue<std::pair<TChannelInfo, TInstant>> UnboundOutputs;
-    bool Connected = false;
     std::weak_ptr<TNodeState> Self;
     // Sender
     ui64 GenMajor = 1;
@@ -623,12 +622,12 @@ public:
     ui64 ReconciliationCount = 0;
     ui64 SeqNo = 0;
     std::atomic<ui64> InflightBytes = 0;
+    NActors::TActorId InputNodeActorId;
     // Receiver
-    NActors::TActorId PeerActorId;
-    std::atomic<ui64> PeerGenMajor = 0;
-    std::atomic<ui64> PeerGenMinor = 0;
+    NActors::TActorId OutputNodeActorId;
+    std::atomic<ui64> OutputNodeGenMajor = 0;
+    std::atomic<ui64> OutputNodeGenMinor = 0;
     ui64 ConfirmedSeqNo = 0;
-    TEvDqCompute::TEvChannelDataV2::TPtr OutOfOrderMessage;
     // ...
     const TDqChannelLimits Limits;
     const ui64 MaxInflightMessages = 8192;
@@ -660,6 +659,10 @@ public:
     std::atomic<bool> ResendAsked = false;
     std::deque<char> ReconciliationLog;
     TChannelInfo LastLostInfo = TChannelInfo(0,  NActors::TActorId{}, NActors::TActorId{});
+    std::atomic<ui64> SendCount = 0;
+    std::atomic<ui64> ResendCount = 0;
+    std::atomic<ui64> ReconCount = 0;
+    std::atomic<TInstant> ReconSent;
 };
 
 class TDebugNodeState : public TNodeState {