Reject requests after getting close session request. (#12618)

* Kqp request can came just after close session request. In this case there is a gap when session actor is already DEAD but kqp proxy stil has a record in the local sessions.

Changelog category:

* Bugfix

Author: dcherednik

ydb/core/kqp/proxy_service/kqp_proxy_service.cpp

@@ -723,6 +723,11 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
         }
 
         if (sessionInfo) {
+            if (sessionInfo->Closing) {
+                TString error = TStringBuilder() << "Session is closing";
+                ReplyProcessError(Ydb::StatusIds::BAD_SESSION, error, requestId);
+                return;
+            }
             LocalSessions->AttachQueryText(sessionInfo, ev->Get()->GetQuery());
         }
 
@@ -784,6 +789,7 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
         }
 
         if (sessionInfo) {
+            LocalSessions->SetSessionClosing(sessionInfo);
             Send(sessionInfo->WorkerId, ev->Release().Release());
         } else {
             if (!sessionId.empty()) {

ydb/core/kqp/proxy_service/kqp_proxy_service_impl.h

@@ -98,7 +98,6 @@ struct TKqpSessionInfo {
     TActorId AttachedRpcId;
     bool PgWire;
     TString QueryText;
-    bool Ready = true;
     TString ClientApplicationName;
     TString ClientSID;
     TString ClientHost;
@@ -111,6 +110,7 @@ struct TKqpSessionInfo {
     TInstant QueryStartAt;
 
     ESessionState State = ESessionState::IDLE;
+    bool Closing = false;
 
     struct TFieldsMap {
         ui64 bitmap = 0;
@@ -246,6 +246,11 @@ class TLocalSessionsRegistry {
         return candidate;
     }
 
+    void SetSessionClosing(const TKqpSessionInfo* sessionInfo) {
+        TKqpSessionInfo* info = const_cast<TKqpSessionInfo*>(sessionInfo);
+        info->Closing = true;
+    }
+
     void StartIdleCheck(const TKqpSessionInfo* sessionInfo, const TDuration idleDuration) {
         if (!sessionInfo) {
             return;