From 22eec508057a224c47dd52ed50c64f7724e1d7e4 Mon Sep 17 00:00:00 2001 From: Dmitry Kardymon Date: Thu, 7 Aug 2025 18:12:29 +0000 Subject: [PATCH 1/3] add FlagTrackDelivery --- .../fq/libs/compute/ydb/status_tracker_actor.cpp | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp b/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp index 6077cdd15534..8fb78cb899d8 100644 --- a/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp +++ b/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp @@ -99,6 +99,7 @@ class TStatusTrackerActor : public TBaseComputeActor { STRICT_STFUNC(StateFunc, hFunc(TEvYdbCompute::TEvGetOperationResponse, Handle); hFunc(TEvents::TEvForwardPingResponse, Handle); + hFunc(NActors::TEvents::TEvUndelivered, Handle); ) void Handle(const TEvents::TEvForwardPingResponse::TPtr& ev) { @@ -231,7 +232,7 @@ class TStatusTrackerActor : public TBaseComputeActor { GetStepCountersSubgroup()->GetCounter("StatIssues", true)->Inc(); } ReportPublicCounters(Builder.PublicStat); - Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest), 0, 1); + Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest), IEventHandle::FlagTrackDelivery, 1); } void UpdateCpuQuota(double cpuUsage) { @@ -254,7 +255,7 @@ class TStatusTrackerActor : public TBaseComputeActor { ReportPublicCounters(Builder.PublicStat); UpdateCpuQuota(Builder.CpuUsage); - Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest)); + Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest), IEventHandle::FlagTrackDelivery); } void Complete() { @@ -270,7 +271,12 @@ class TStatusTrackerActor : public TBaseComputeActor { ReportPublicCounters(Builder.PublicStat); UpdateCpuQuota(Builder.CpuUsage); - Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest)); + Send(Pinger, new TEvents::TEvForwardPingRequest(pingTaskRequest), IEventHandle::FlagTrackDelivery); + } + + void Handle(NActors::TEvents::TEvUndelivered::TPtr& ev) { + LOG_W("TEvUndelivered, from " << ev->Sender << ", reason " << ev->Get()->Reason); + FailedAndPassAway(); } private: From 9892962013b53c1d727b7649c48fbd35ba5f2d9c Mon Sep 17 00:00:00 2001 From: Dmitry Kardymon Date: Thu, 7 Aug 2025 19:25:42 +0000 Subject: [PATCH 2/3] add comment --- ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp b/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp index 8fb78cb899d8..2d37375db464 100644 --- a/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp +++ b/ydb/core/fq/libs/compute/ydb/status_tracker_actor.cpp @@ -275,6 +275,8 @@ class TStatusTrackerActor : public TBaseComputeActor { } void Handle(NActors::TEvents::TEvUndelivered::TPtr& ev) { + // Pinger may end (with ydb_run_actor); no one sends a poison pill to this status tracker. + // Therefore we'll finish ourselves. LOG_W("TEvUndelivered, from " << ev->Sender << ", reason " << ev->Get()->Reason); FailedAndPassAway(); } From bacb1afc8edc60baca4953c52074e1c5ed45772f Mon Sep 17 00:00:00 2001 From: Dmitry Kardymon Date: Fri, 8 Aug 2025 12:02:53 +0000 Subject: [PATCH 3/3] fix IsSuccess --- .../fq/libs/control_plane_storage/ydb_control_plane_storage.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ydb/core/fq/libs/control_plane_storage/ydb_control_plane_storage.cpp b/ydb/core/fq/libs/control_plane_storage/ydb_control_plane_storage.cpp index 62f89f27ca6b..0ceac0cef108 100644 --- a/ydb/core/fq/libs/control_plane_storage/ydb_control_plane_storage.cpp +++ b/ydb/core/fq/libs/control_plane_storage/ydb_control_plane_storage.cpp @@ -527,7 +527,7 @@ NThreading::TFuture TYdbControlPlaneStorageActor::PickTask( taskParams.PrepareParams, requestCounters, debugInfo, validators, transactionMode, taskParams.RetryOnTli) .Apply([=, responseTasks=responseTasks, queryId = taskParams.QueryId](const auto& future) { const auto status = future.GetValue(); - if (responseTasks && status.GetStatus() == EStatus::GENERIC_ERROR) { + if (responseTasks && !status.IsSuccess()) { responseTasks->SafeEraseTaskBlocking(queryId); } });