Merge secrets fixes to stable-25-3-1 (#26705)

Author: yurikiselev

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -2263,7 +2263,7 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
             TActorSetupCmd(finalize, TMailboxType::HTSwap, appData->UserPoolId)));
 
         if (appData->FeatureFlags.GetEnableSchemaSecrets()) {
-            auto describeSchemaSecretsService = NKqp::CreateDescribeSchemaSecretsService();
+            auto describeSchemaSecretsService = NKqp::TDescribeSchemaSecretsServiceFactory().CreateService();
             setup->LocalServices.push_back(std::make_pair(
                 NKqp::MakeKqpDescribeSchemaSecretServiceId(NodeId),
                 TActorSetupCmd(describeSchemaSecretsService, TMailboxType::HTSwap, appData->UserPoolId)));

ydb/core/kqp/common/events/script_executions.h

@@ -3,6 +3,7 @@
 #include <ydb/core/protos/kqp.pb.h>
 #include <ydb/core/protos/kqp_stats.pb.h>
 #include <ydb/core/protos/kqp_physical.pb.h>
+#include <ydb/library/aclib/aclib.h>
 #include <yql/essentials/public/issue/yql_issue.h>
 #include <ydb/public/api/protos/ydb_operation.pb.h>
 #include <ydb/public/api/protos/ydb_query.pb.h>
@@ -281,7 +282,7 @@ struct TEvSaveScriptExternalEffectRequest : public TEventLocal<TEvSaveScriptExte
         TString Database;
 
         TString CustomerSuppliedId;
-        TString UserToken;
+        TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
         std::vector<NKqpProto::TKqpExternalSink> Sinks;
         std::vector<TString> SecretNames;
     };
@@ -386,7 +387,7 @@ struct TEvSaveScriptFinalStatusResponse : public TEventLocal<TEvSaveScriptFinalS
     bool OperationAlreadyFinalized = false;
     bool WaitRetry = false;
     TString CustomerSuppliedId;
-    TString UserToken;
+    TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
     std::vector<NKqpProto::TKqpExternalSink> Sinks;
     std::vector<TString> SecretNames;
     Ydb::StatusIds::StatusCode Status;

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -946,7 +946,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
     }
 
     void GetSecretsSnapshot() {
-        RegisterDescribeSecretsActor(this->SelfId(), UserToken ? UserToken->GetUserSID() : "", SecretNames, this->ActorContext().ActorSystem());
+        RegisterDescribeSecretsActor(this->SelfId(), UserToken, Database, SecretNames, this->ActorContext().ActorSystem());
     }
 
     void GetResourcesSnapshot() {

ydb/core/kqp/federated_query/kqp_federated_query_actors.cpp

@@ -2,15 +2,16 @@
 
 #include <ydb/core/kqp/common/events/script_executions.h>
 #include <ydb/core/protos/flat_scheme_op.pb.h>
+#include <ydb/core/tx/scheme_cache/scheme_cache.h>
+#include <ydb/core/tx/tx_proxy/proxy.h>
+#include <ydb/core/tx/schemeshard/schemeshard.h>
+#include <ydb/core/tx/scheme_board/events.h>
 
 #include <ydb/library/actors/core/actor.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/aclib/aclib.h>
-#include <library/cpp/threading/future/future.h>
 
-#include <ydb/core/tx/scheme_cache/scheme_cache.h>
-#include <ydb/core/tx/tx_proxy/proxy.h>
-#include <ydb/core/tx/schemeshard/schemeshard.h>
+#include <library/cpp/threading/future/future.h>
 
 namespace NKikimr::NKqp {
 
@@ -24,60 +25,127 @@ class TDescribeSchemaSecretsService: public NActors::TActorBootstrapped<TDescrib
     struct TEvResolveSecret : public NActors::TEventLocal<TEvResolveSecret, EvResolveSecret> {
     public:
         TEvResolveSecret(
-            const TString& ownerUserId,
-            const TString& secretName,
+            const TIntrusiveConstPtr<NACLib::TUserToken> userToken,
+            const TString& database,
+            const TVector<TString>& secretNames,
             NThreading::TPromise<TEvDescribeSecretsResponse::TDescription> promise
         )
-            : UserToken(NACLib::TUserToken{ownerUserId, TVector<NACLib::TSID>{}})
-            , SecretName(secretName)
+            : UserToken(userToken)
+            , Database(database)
+            , SecretNames(secretNames)
             , Promise(promise)
         {
+            Y_ENSURE(!Database.empty(), "Database name must be set in secret requests");
         }
 
     public:
-        const NACLib::TUserToken UserToken;
-        const TString SecretName;
+        const TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+        const TString Database;
+        const TVector<TString> SecretNames;
         NThreading::TPromise<TEvDescribeSecretsResponse::TDescription> Promise;
     };
 
+private:
+    struct TVersionedSecret {
+        ui64 SecretVersion = 0;
+        ui64 PathId = 0;
+        TString Name;
+        TString Value;
+    };
+
+    struct TResponseContext {
+        using TIncomingOrderId = ui64;
+        THashMap<TString, TIncomingOrderId> Secrets;
+        NThreading::TPromise<TEvDescribeSecretsResponse::TDescription> Result;
+        size_t FilledSecretsCnt = 0;
+    };
+
 private:
     STRICT_STFUNC(StateWait,
-        hFunc(TEvResolveSecret, Handle);
-        hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
-        hFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle);
+        hFunc(TEvResolveSecret, HandleIncomingRequest);
+        hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleSchemeCacheResponse);
+        hFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, HandleSchemeShardResponse);
+        hFunc(TSchemeBoardEvents::TEvNotifyDelete, HandleNotifyDelete);
+        hFunc(TSchemeBoardEvents::TEvNotifyUpdate, HandleNotifyUpdate);
         cFunc(NActors::TEvents::TEvPoison::EventType, PassAway);
     )
 
-    void Handle(TEvResolveSecret::TPtr& ev);
-    void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
-    void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev);
-    void FillResponse(const ui64 requestId, const TEvDescribeSecretsResponse::TDescription& response);
-    void SaveIncomingRequestInfo(const TEvResolveSecret& req);
-    void SendSchemeCacheRequest(const TString& secretName);
+    void HandleIncomingRequest(TEvResolveSecret::TPtr& ev);
+    void HandleSchemeCacheResponse(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
+    void HandleSchemeShardResponse(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev);
+    void HandleNotifyDelete(TSchemeBoardEvents::TEvNotifyDelete::TPtr& ev);
+    void HandleNotifyUpdate(TSchemeBoardEvents::TEvNotifyUpdate::TPtr& ev);
+
+    void FillResponse(const ui64& requestId, const TEvDescribeSecretsResponse::TDescription& response);
+    void SaveIncomingRequestInfo(const TEvResolveSecret& ev);
+    void SendSchemeCacheRequests(const TEvResolveSecret& ev);
+    bool LocalCacheHasActualVersion(const TVersionedSecret& secret, const ui64& cacheSecretVersion);
+    bool LocalCacheHasActualObject(const TVersionedSecret& secret, const ui64& cacheSecretPathId);
+    bool HandleSchemeCacheErrorsIfAny(const ui64& requestId, NSchemeCache::TSchemeCacheNavigate& result);
+    void FillResponseIfFinished(const ui64& requestId, const TResponseContext& responseCtx);
 
 public:
     TDescribeSchemaSecretsService() = default;
 
     void Bootstrap();
 
-private:
-    struct TVersionedSecret {
-        ui64 Version;
-        TString Value;
+public:
+    // For tests only
+    class ISecretUpdateListener : public TThrRefBase {
+    public:
+        virtual void HandleNotifyDelete(const TString& secretName) = 0;
+        virtual ~ISecretUpdateListener() = default;
     };
+    void SetSecretUpdateListener(ISecretUpdateListener* secretUpdateListener) {
+        SecretUpdateListener = secretUpdateListener;
+    }
 
+private:
     ui64 LastCookie = 0;
-    THashMap<ui64, NThreading::TPromise<TEvDescribeSecretsResponse::TDescription>> ResolveInFlight;
-    THashMap<ui64, TString> SecretNameInFlight;
-    THashMap<TString, TVersionedSecret> SecretNameToValue;
+    THashMap<ui64, TResponseContext> ResolveInFlight;
+    THashMap<TString, TVersionedSecret> VersionedSecrets;
+    THashMap<TString, TActorId> SchemeBoardSubscribers;
+    ISecretUpdateListener* SecretUpdateListener;
 };
 
-IActor* CreateDescribeSecretsActor(const TString& ownerUserId, const std::vector<TString>& secretIds, NThreading::TPromise<TEvDescribeSecretsResponse::TDescription> promise);
+void RegisterDescribeSecretsActor(
+    const NActors::TActorId& replyActorId,
+    const TIntrusiveConstPtr<NACLib::TUserToken> userToken,
+    const TString& database,
+    const std::vector<TString>& secretIds,
+    NActors::TActorSystem* actorSystem
+);
+
+NThreading::TFuture<TEvDescribeSecretsResponse::TDescription> DescribeExternalDataSourceSecrets(
+    const NKikimrSchemeOp::TAuth& authDescription,
+    const TIntrusiveConstPtr<NACLib::TUserToken> userToken,
+    const TString& database,
+    TActorSystem* actorSystem
+);
 
-void RegisterDescribeSecretsActor(const TActorId& replyActorId, const TString& ownerUserId, const std::vector<TString>& secretIds, TActorSystem* actorSystem);
+IActor* CreateDescribeSchemaSecretsService();
 
-NThreading::TFuture<TEvDescribeSecretsResponse::TDescription> DescribeExternalDataSourceSecrets(const NKikimrSchemeOp::TAuth& authDescription, const TString& ownerUserId, TActorSystem* actorSystem);
+class IDescribeSchemaSecretsServiceFactory {
+public:
+    using TPtr = std::shared_ptr<IDescribeSchemaSecretsServiceFactory>;
 
-IActor* CreateDescribeSchemaSecretsService();
+    virtual IActor* CreateService() = 0;
+    virtual ~IDescribeSchemaSecretsServiceFactory() = default;
+};
+
+class TDescribeSchemaSecretsServiceFactory : public IDescribeSchemaSecretsServiceFactory {
+public:
+    IActor* CreateService() override;
+};
+
+NThreading::TFuture<TEvDescribeSecretsResponse::TDescription> DescribeSecret(
+    const TVector<TString>& secretNames,
+    const TIntrusiveConstPtr<NACLib::TUserToken> userToken,
+    const TString& database,
+    TActorSystem* actorSystem
+);
+
+bool UseSchemaSecrets(const NKikimr::TFeatureFlags& flags, const TVector<TString>& secretNames);
+bool UseSchemaSecrets(const NKikimr::TFeatureFlags& flags, const TString& secretName);
 
 }  // namespace NKikimr::NKqp