provide database in schemecache request (#15940)

Provide database name in schemecache request.
Without database name schemecache could answer with PathUnknown instead of Unavailable in some cases.
This leads to bug, described in KIKIMR-22718.

Author: alexnick88

ydb/core/client/server/msgbus_server_pq_metacache.cpp

@@ -519,7 +519,11 @@ class TPersQueueMetaCacheActor : public TActorBootstrapped<TPersQueueMetaCacheAc
         auto inserted = DescribeTopicsWaiters.insert(std::make_pair(reqId, waiter)).second;
         Y_ABORT_UNLESS(inserted);
 
+        TMaybe<TString> db = {};
+
         for (const auto& [path, database] : waiter->GetTopics()) {
+            if (!db) db = database;
+            if (*db != database) db = "";
             auto split = NKikimr::SplitPath(path);
             TSchemeCacheNavigate::TEntry entry;
             if (!split.empty()) {
@@ -532,8 +536,9 @@ class TPersQueueMetaCacheActor : public TActorBootstrapped<TPersQueueMetaCacheAc
 
             schemeCacheRequest->ResultSet.emplace_back(std::move(entry));
         }
-
-        LOG_DEBUG_S(ctx, NKikimrServices::PQ_METACACHE, "send request for " << (waiter->Type == EWaiterType::DescribeAllTopics ? " all " : "") << waiter->GetTopics().size() << " topics, got " << DescribeTopicsWaiters.size() << " requests infly");
+        if (db) schemeCacheRequest->DatabaseName = *db;
+        LOG_DEBUG_S(ctx, NKikimrServices::PQ_METACACHE, "send request for " << (waiter->Type == EWaiterType::DescribeAllTopics ? " all " : "")
+                             << waiter->GetTopics().size() << " topics, got " << DescribeTopicsWaiters.size() << " requests infly, db = \"" << db << "\"");
 
         ctx.Send(SchemeCacheId, new TEvTxProxySchemeCache::TEvNavigateKeySet(schemeCacheRequest.release()), 0, 0, waiter->Span.GetTraceId());
     }

ydb/core/http_proxy/http_req.cpp

@@ -1458,6 +1458,7 @@ namespace NKikimr::NHttpProxy {
             entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpPath;
             entry.SyncVersion = false;
             schemeCacheRequest->ResultSet.emplace_back(entry);
+            schemeCacheRequest->DatabaseName = CanonizePath(DatabasePath);
             ctx.Send(MakeSchemeCacheID(), MakeHolder<TEvTxProxySchemeCache::TEvNavigateKeySet>(schemeCacheRequest.release()));
         }
 

ydb/core/kafka_proxy/actors/kafka_produce_actor.cpp

@@ -558,6 +558,8 @@ void TKafkaProduceActor::ProcessInitializationRequests(const TActorContext& ctx)
         request->ResultSet.emplace_back(entry);
     }
 
+    request->DatabaseName = CanonizePath(Context->DatabasePath);
+
     ctx.Send(MakeSchemeCacheID(), MakeHolder<TEvTxProxySchemeCache::TEvNavigateKeySet>(request.release()));
 }
 

ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp

@@ -182,13 +182,25 @@ void TKafkaSaslAuthActor::SendDescribeRequest(const TActorContext& ctx) {
     entry.Operation = NKikimr::NSchemeCache::TSchemeCacheNavigate::OpPath;
     entry.SyncVersion = false;
     schemeCacheRequest->ResultSet.emplace_back(entry);
+    schemeCacheRequest->DatabaseName = CanonizePath(DatabasePath);
     ctx.Send(NKikimr::MakeSchemeCacheID(), MakeHolder<NKikimr::TEvTxProxySchemeCache::TEvNavigateKeySet>(schemeCacheRequest.release()));
 }
 
 void TKafkaSaslAuthActor::Handle(NKikimr::TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
     const NKikimr::NSchemeCache::TSchemeCacheNavigate* navigate = ev->Get()->Request.Get();
     if (navigate->ErrorCount) {
-        SendResponseAndDie(EKafkaErrors::SASL_AUTHENTICATION_FAILED, "", TStringBuilder() << "Database with path '" << DatabasePath << "' doesn't exists", ctx);
+        switch(navigate->ResultSet.front().Status) {
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::RootUnknown:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::PathNotTable:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::PathNotPath:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::TableCreationNotComplete:
+                return SendResponseAndDie(EKafkaErrors::SASL_AUTHENTICATION_FAILED, "", TStringBuilder() << "Database with path '" << DatabasePath << "' doesn't exists", ctx);
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::AccessDenied:
+                return SendResponseAndDie(EKafkaErrors::SASL_AUTHENTICATION_FAILED, "", TStringBuilder() << "Database with path '" << DatabasePath << "' access denied", ctx);
+            default:
+                return SendResponseAndDie(EKafkaErrors::BROKER_NOT_AVAILABLE, "", TStringBuilder() << "Internal error with navigate status " << navigate->ResultSet.front().Status, ctx);
+        }
         return;
     }
     Y_ABORT_UNLESS(navigate->ResultSet.size() == 1);

ydb/services/datastreams/put_records_actor.h

@@ -312,6 +312,7 @@ namespace NKikimr::NDataStreams::V1 {
         entry.Path = NKikimr::SplitPath(this->GetTopicPath());
         entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
         entry.SyncVersion = true;
+        schemeCacheRequest->DatabaseName = CanonizePath(this->Request_->GetDatabaseName().GetOrElse(""));
         schemeCacheRequest->ResultSet.emplace_back(entry);
         ctx.Send(MakeSchemeCacheID(), MakeHolder<TEvTxProxySchemeCache::TEvNavigateKeySet>(schemeCacheRequest.release()));
     }

ydb/services/persqueue_v1/actors/read_init_auth_actor.cpp

@@ -84,6 +84,7 @@ void TReadInitAndAuthActor::SendCacheNavigateRequest(const TActorContext& ctx, c
     entry.SyncVersion = true;
     entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpPath;
     schemeCacheRequest->ResultSet.emplace_back(entry);
+    schemeCacheRequest->DatabaseName = AppData(ctx)->PQConfig.GetDatabase();
     LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " Send client acl request");
     ctx.Send(NewSchemeCache, new TEvTxProxySchemeCache::TEvNavigateKeySet(schemeCacheRequest.Release()));
 }

ydb/services/persqueue_v1/persqueue_compat_ut.cpp

@@ -42,10 +42,20 @@ class TPQv1CompatTestBase {
         Server->AnnoyingClient->CreateTopicNoLegacy(
                 "/Root/LbCommunal/account/topic2", 1, true, true, {}, {}, "account"
         );
-
         Server->AnnoyingClient->CreateTopicNoLegacy(
                 "/Root/LbCommunal/account/topic2-mirrored-from-dc2", 1, true, false, {}, {}, "account"
         );
+
+        Server->AnnoyingClient->MkDir("/Root", "LbCommunal");
+        Server->AnnoyingClient->MkDir("/Root/LbCommunal", "account2");
+        Server->AnnoyingClient->CreateTopicNoLegacy(
+                "/Root/LbCommunal/account2/topic3", 1, true, true, {}, {}, "account2"
+        );
+        Server->AnnoyingClient->CreateTopicNoLegacy(
+                "/Root/LbCommunal/account2/topic3-mirrored-from-dc2", 1, true, false, {}, {}, "account2"
+        );
+
+
         Server->AnnoyingClient->CreateConsumer("test-consumer");
         InitPQLib();
     }
@@ -153,6 +163,12 @@ Y_UNIT_TEST_SUITE(TPQCompatTest) {
             GetLocks({"account/topic1", "account/topic2"}, rs);
             rs->Close();
         }
+        {
+            auto rs = testServer.CreateReadSession({"account/topic1", "account/topic2", "account2/topic3"});
+            GetLocks({"account/topic1", "account/topic2", "account2/topic3"}, rs);
+            rs->Close();
+        }
+
     }
 
     Y_UNIT_TEST(BadTopics) {