YQ-3600 fixed sessions leak in metadata service (#8312)

Author: GrigoriyPA

ydb/services/metadata/ds_table/accessor_refresh.cpp

@@ -9,6 +9,7 @@ void TDSAccessorRefresher::OnBootstrap() {
 }
 
 void TDSAccessorRefresher::OnNewEnrichedSnapshot(NFetcher::ISnapshot::TPtr snapshot) {
+    FetchingRequestIsRunning = false;
     Schedule(Config.GetRefreshPeriod(), new TEvRefresh());
     CurrentSnapshot = snapshot;
     *CurrentSelection.mutable_result_sets() = std::move(*ProposedProto.mutable_result_sets());
@@ -22,19 +23,24 @@ void TDSAccessorRefresher::OnNewParsedSnapshot(Ydb::Table::ExecuteQueryResult&&
         ALS_INFO(NKikimrServices::METADATA_PROVIDER) << "New refresher data: " << ProposedProto.DebugString();
         SnapshotConstructor->EnrichSnapshotData(snapshot, InternalController);
     } else {
+        FetchingRequestIsRunning = false;
         CurrentSnapshot->SetActuality(GetRequestedActuality());
         OnSnapshotRefresh();
         Schedule(Config.GetRefreshPeriod(), new TEvRefresh());
     }
 }
 
 void TDSAccessorRefresher::OnConstructSnapshotError(const TString& errorMessage) {
+    FetchingRequestIsRunning = false;
     TBase::OnConstructSnapshotError(errorMessage);
     Schedule(Config.GetRefreshPeriod(), new TEvRefresh());
 }
 
 void TDSAccessorRefresher::Handle(TEvRefresh::TPtr& /*ev*/) {
-    TBase::StartSnapshotsFetching();
+    if (!FetchingRequestIsRunning) {
+        FetchingRequestIsRunning = true;
+        TBase::StartSnapshotsFetching();
+    }
 }
 
 }

ydb/services/metadata/ds_table/accessor_refresh.h

@@ -9,6 +9,7 @@ class TDSAccessorRefresher: public TDSAccessorBase {
     YDB_READONLY_DEF(NFetcher::ISnapshot::TPtr, CurrentSnapshot);
     YDB_READONLY_DEF(Ydb::Table::ExecuteQueryResult, CurrentSelection);
     Ydb::Table::ExecuteQueryResult ProposedProto;
+    bool FetchingRequestIsRunning = false;
     const TConfig Config;
 
     void Handle(TEvRefresh::TPtr& ev);

ydb/services/metadata/ds_table/accessor_subscribe.cpp

@@ -52,6 +52,9 @@ void TDSAccessorNotifier::OnSnapshotRefresh() {
             ++it;
         }
     }
+    if (!Asked.empty()) {
+        Sender<TEvRefresh>().SendTo(SelfId());
+    }
 }
 
 void TDSAccessorNotifier::OnBootstrap() {

ydb/services/metadata/manager/alter_impl.h

@@ -120,10 +120,10 @@ class TModificationActorImpl: public NActors::TActorBootstrapped<TModificationAc
         InitState();
         if (!Patches.size()) {
             ExternalController->OnAlteringProblem("no patches");
-            return TBase::PassAway();
+            return this->PassAway();
         }
         if (!BuildRestoreObjectIds()) {
-            return TBase::PassAway();
+            return this->PassAway();
         }
 
         TBase::Register(new NRequest::TYDBCallbackRequest<NRequest::TDialogCreateSession>(
@@ -146,7 +146,7 @@ class TModificationActorImpl: public NActors::TActorBootstrapped<TModificationAc
         Y_ABORT_UNLESS(TransactionId);
         std::vector<TObject> objects = std::move(ev->Get()->MutableObjects());
         if (!PrepareRestoredObjects(objects)) {
-            TBase::PassAway();
+            this->PassAway();
         } else {
             Manager->PrepareObjectsBeforeModification(std::move(objects), InternalController, Context, TAlterOperationContext(SessionId, TransactionId, RestoreObjectIds));
         }
@@ -159,12 +159,12 @@ class TModificationActorImpl: public NActors::TActorBootstrapped<TModificationAc
         for (auto&& i : ev->Get()->GetObjects()) {
             if (!records.AddRecordNativeValues(i.SerializeToRecord())) {
                 ExternalController->OnAlteringProblem("unexpected serialization inconsistency");
-                return TBase::PassAway();
+                return this->PassAway();
             }
         }
         if (!ProcessPreparedObjects(std::move(records))) {
             ExternalController->OnAlteringProblem("cannot process prepared objects");
-            return TBase::PassAway();
+            return this->PassAway();
         }
     }
 
@@ -183,6 +183,15 @@ class TModificationActorImpl: public NActors::TActorBootstrapped<TModificationAc
         ExternalController->OnAlteringProblem("cannot restore objects: " + ev->Get()->GetErrorMessage());
     }
 
+    void PassAway() override {
+        if (SessionId) {
+            NMetadata::NRequest::TDialogDeleteSession::TRequest deleteRequest;
+            deleteRequest.set_session_id(SessionId);
+            TBase::Register(new NRequest::TYDBCallbackRequest<NRequest::TDialogDeleteSession>(deleteRequest, UserToken, TBase::SelfId()));
+        }
+
+        TBase::PassAway();
+    }
 };
 
 template <class TObject>