Run deferred publish finalize KQP ops as metadata@system (#47232)

Co-authored-by: Alexander Kotov <[email protected]>

4 files changed, 11 insertions, 63 deletions

diff --git a/ydb/core/grpc_services/rpc_topic_deferred_publish.cpp b/ydb/core/grpc_services/rpc_topic_deferred_publish.cpp
index 9f7af45356a..7af2113867d 100644
--- a/ydb/core/grpc_services/rpc_topic_deferred_publish.cpp
+++ b/ydb/core/grpc_services/rpc_topic_deferred_publish.cpp
@@ -23,13 +23,6 @@ constexpr TStringBuf NotImplementedMessage = "Topic deferred publish is not impl
 constexpr TStringBuf DisabledMessage = "Topic deferred publish is not enabled";
 constexpr TStringBuf AuthenticationRequiredMessage = "Authentication is required";
 
-TString GetSerializedUserToken(const IRequestOpCtx* request) {
-    if (request == nullptr) {
-        return {};
-    }
-    return request->GetSerializedToken();
-}
-
 TString GetUserSID(const IRequestOpCtx* request) {
     if (request == nullptr) {
         return BUILTIN_ACL_NO_USER_SID;
@@ -423,7 +416,6 @@ public:
             *database,
             protoRequest->int_publication_id(),
             NPQ::NDeferredPublish::EFinalizePublicationOp::Publish,
-            GetSerializedUserToken(Request.get()),
             callerSid));
         Become(&TPublishRequestActor::StateFunc);
     }
@@ -492,7 +484,6 @@ public:
             *database,
             protoRequest->int_publication_id(),
             NPQ::NDeferredPublish::EFinalizePublicationOp::Cancel,
-            GetSerializedUserToken(Request.get()),
             callerSid));
         Become(&TCancelPublicationRequestActor::StateFunc);
     }
diff --git a/ydb/core/persqueue/deferred_publish/finalize_publication_actor.cpp b/ydb/core/persqueue/deferred_publish/finalize_publication_actor.cpp
index 5890aa4f953..713fd40ff7a 100644
--- a/ydb/core/persqueue/deferred_publish/finalize_publication_actor.cpp
+++ b/ydb/core/persqueue/deferred_publish/finalize_publication_actor.cpp
@@ -21,6 +21,10 @@ namespace {
 
 using namespace NKikimr::NKqp;
 
+TString MetadataUserToken() {
+    return NACLib::TSystemUsers::Metadata().SerializeAsString();
+}
+
 TString MakeDeletePublicationSql() {
     return TStringBuilder() << R"(
         -- TFinalizePublicationActor::MakeDeletePublicationSql
@@ -111,13 +115,11 @@ public:
         TString database,
         ui64 intPublicationId,
         EFinalizePublicationOp op,
-        TString userToken,
         TString callerSid)
         : ReplyTo(replyTo)
         , Database(std::move(database))
         , IntPublicationId(intPublicationId)
         , Op(op)
-        , UserToken(std::move(userToken))
         , CallerSid(std::move(callerSid))
     {}
 
@@ -158,19 +160,11 @@ private:
         Become(&TFinalizePublicationActor::StateDeleteOnly);
     }
 
-    void SetCreateSessionIdentity(TEvKqp::TEvCreateSessionRequest& ev) {
-        if (!UserToken.empty()) {
-            NACLibProto::TUserToken token;
-            if (token.ParseFromString(UserToken)) {
-                ev.Record.SetUserSID(token.GetUserSID());
-            }
-        }
-    }
-
     void StartKqpSession() {
         auto ev = MakeHolder<TEvKqp::TEvCreateSessionRequest>();
         ev->Record.MutableRequest()->SetDatabase(Database);
-        SetCreateSessionIdentity(*ev);
+        // Registry and topic finalize ops run as metadata@system, like Begin/List query actors.
+        ev->Record.SetUserSID(BUILTIN_ACL_METADATA);
         Send(MakeKqpProxyID(SelfId().NodeId()), ev.Release());
         Step = EStep::KqpCreateSession;
         Become(&TFinalizePublicationActor::StateKqp);
@@ -178,9 +172,7 @@ private:
 
     void SendKqpBeginTx() {
         auto ev = MakeHolder<TEvKqp::TEvQueryRequest>();
-        if (!UserToken.empty()) {
-            ev->Record.SetUserToken(UserToken);
-        }
+        ev->Record.SetUserToken(MetadataUserToken());
         ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_BEGIN_TX);
         ev->Record.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
         ev->Record.MutableRequest()->SetSessionId(KqpSessionId);
@@ -197,9 +189,7 @@ private:
                 .Build();
 
         auto ev = MakeHolder<TEvKqp::TEvQueryRequest>();
-        if (!UserToken.empty()) {
-            ev->Record.SetUserToken(UserToken);
-        }
+        ev->Record.SetUserToken(MetadataUserToken());
         ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
         ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
         ev->Record.MutableRequest()->SetQuery(MakeDeletePublicationSql());
@@ -218,9 +208,7 @@ private:
 
     void SendKqpDeferredPublication() {
         auto ev = MakeHolder<TEvKqp::TEvQueryRequest>();
-        if (!UserToken.empty()) {
-            ev->Record.SetUserToken(UserToken);
-        }
+        ev->Record.SetUserToken(MetadataUserToken());
         ev->Record.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_UNDEFINED);
         ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_TOPIC);
         ev->Record.MutableRequest()->SetDatabase(Database);
@@ -235,9 +223,7 @@ private:
 
     void SendKqpCommit() {
         auto ev = MakeHolder<TEvKqp::TEvQueryRequest>();
-        if (!UserToken.empty()) {
-            ev->Record.SetUserToken(UserToken);
-        }
+        ev->Record.SetUserToken(MetadataUserToken());
         ev->Record.MutableRequest()->SetAction(NKikimrKqp::QUERY_ACTION_COMMIT_TX);
         ev->Record.MutableRequest()->MutableTxControl()->set_tx_id(TxId);
         ev->Record.MutableRequest()->MutableTxControl()->set_commit_tx(true);
@@ -369,7 +355,6 @@ private:
     const TString Database;
     const ui64 IntPublicationId;
     const EFinalizePublicationOp Op;
-    const TString UserToken;
     const TString CallerSid;
 
     TMaybe<TListDestinationsData> ListDestinationsData;
@@ -386,10 +371,9 @@ NActors::IActor* CreateFinalizePublicationActor(
     const TString& database,
     ui64 intPublicationId,
     EFinalizePublicationOp op,
-    const TString& userToken,
     const TString& callerSid)
 {
-    return new TFinalizePublicationActor(replyTo, database, intPublicationId, op, userToken, callerSid);
+    return new TFinalizePublicationActor(replyTo, database, intPublicationId, op, callerSid);
 }
 
 } // namespace NKikimr::NPQ::NDeferredPublish
diff --git a/ydb/core/persqueue/deferred_publish/finalize_publication_actor.h b/ydb/core/persqueue/deferred_publish/finalize_publication_actor.h
index b14130cd4e4..28c33b38dd2 100644
--- a/ydb/core/persqueue/deferred_publish/finalize_publication_actor.h
+++ b/ydb/core/persqueue/deferred_publish/finalize_publication_actor.h
@@ -11,7 +11,6 @@ NActors::IActor* CreateFinalizePublicationActor(
     const TString& database,
     ui64 intPublicationId,
     EFinalizePublicationOp op,
-    const TString& userToken,
     const TString& callerSid);
 
 } // namespace NKikimr::NPQ::NDeferredPublish
diff --git a/ydb/services/persqueue_v1/ut/topic_deferred_publish_ut.cpp b/ydb/services/persqueue_v1/ut/topic_deferred_publish_ut.cpp
index cfbe9e4c964..686ffd7eb56 100644
--- a/ydb/services/persqueue_v1/ut/topic_deferred_publish_ut.cpp
+++ b/ydb/services/persqueue_v1/ut/topic_deferred_publish_ut.cpp
@@ -431,17 +431,6 @@ void GrantPublicationTableWrite(NPersQueue::TTestServer& server, const TString&
     }
 }
 
-void GrantPublicationRegistryDelete(NPersQueue::TTestServer& server, const TString& subject) {
-    GrantPublicationTableWrite(server, subject);
-    if (SchemePathExists(server, "/Root/.metadata/topic_deferred_publications")) {
-        server.AnnoyingClient->TestGrant(
-            "/Root/.metadata",
-            "topic_deferred_publications",
-            subject,
-            NACLib::EAccessRights::GenericWrite);
-    }
-}
-
 void InsertDestinationRow(
     NPersQueue::TTestServer& server,
     const TString& authTicket,
@@ -2035,7 +2024,6 @@ Y_UNIT_TEST_SUITE(TopicDeferredPublishFinalize) {
 
 Y_UNIT_TEST(PublishAfterStreamWriteClearsRegistryAndMakesDataVisible) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-publish-topic", "ext-publish");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     constexpr TStringBuf payload = "deferred-payload-visible";
     {
@@ -2059,7 +2047,6 @@ Y_UNIT_TEST(PublishAfterStreamWriteClearsRegistryAndMakesDataVisible) {
 
 Y_UNIT_TEST(PublishAfterStreamWriteToTwoPartitionsMakesDataVisible) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-two-partitions-topic", "ext-two-partitions");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     constexpr TStringBuf payload0 = "deferred-payload-part-0";
     constexpr TStringBuf payload1 = "deferred-payload-part-1";
@@ -2094,7 +2081,6 @@ Y_UNIT_TEST(PublishAfterStreamWriteToTwoPartitionsMakesDataVisible) {
 
 Y_UNIT_TEST(CancelAfterStreamWriteToTwoPartitionsClearsRegistryWithoutData) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-cancel-two-partitions-topic", "ext-cancel-two-partitions");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     {
         auto session = fixture.OpenWriteStream("producer-cancel-0", 0);
@@ -2123,7 +2109,6 @@ Y_UNIT_TEST(CancelAfterStreamWriteToTwoPartitionsClearsRegistryWithoutData) {
 
 Y_UNIT_TEST(CancelAfterStreamWriteClearsRegistryWithoutData) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-cancel-topic", "ext-cancel");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     {
         auto session = fixture.OpenWriteStream("producer-cancel");
@@ -2144,7 +2129,6 @@ Y_UNIT_TEST(CancelAfterStreamWriteClearsRegistryWithoutData) {
 
 Y_UNIT_TEST(RepeatFinalizeReturnsNotFound) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-repeat-topic", "ext-repeat");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     {
         auto session = fixture.OpenWriteStream("producer-repeat");
@@ -2209,7 +2193,6 @@ Y_UNIT_TEST(PublishMultipleDestinations) {
     auto topicStub = MakeTopicServiceStub(server);
     const ui64 intPublicationId = BeginPublicationIntId(
         CallBeginPublication(*deferredStub, "/Root", "ext-multi"));
-    GrantPublicationRegistryDelete(server, "root@builtin");
 
     {
         auto session = TStreamWriteSession::Open(*topicStub, "finalize-multi-topic-a", "producer-a", 0);
@@ -2233,7 +2216,6 @@ Y_UNIT_TEST(PublishMultipleDestinations) {
 
 Y_UNIT_TEST(PublishBeforeWriteAckKeepsRegistry) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-before-ack-topic", "ext-before-ack");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     TFinalizePublicationOutcome publishOutcome;
     std::thread publishThread([&]() {
@@ -2253,7 +2235,6 @@ Y_UNIT_TEST(PublishBeforeWriteAckKeepsRegistry) {
 
 Y_UNIT_TEST(PublishFailureOnInvalidDestinationKeepsRegistry) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("finalize-bad-dest-topic", "ext-bad-dest");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     {
         auto session = fixture.OpenWriteStream("producer-bad-dest");
@@ -2303,7 +2284,6 @@ Y_UNIT_TEST_SUITE(TopicDeferredPublishLifecycle) {
 
 Y_UNIT_TEST(PublishMakesDataVisible) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("lifecycle-publish-topic", "ext-lifecycle-publish");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     constexpr TStringBuf payload = "lifecycle-publish-payload";
     {
@@ -2327,7 +2307,6 @@ Y_UNIT_TEST(PublishMakesDataVisible) {
 
 Y_UNIT_TEST(CancelDiscardsData) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("lifecycle-cancel-topic", "ext-lifecycle-cancel");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     {
         auto session = fixture.OpenWriteStream("producer-lifecycle-cancel");
@@ -2346,7 +2325,6 @@ Y_UNIT_TEST(CancelDiscardsData) {
 
 Y_UNIT_TEST(StagingNotVisibleBeforePublish) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("lifecycle-staging-topic", "ext-lifecycle-staging");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     constexpr TStringBuf payload = "lifecycle-staging-payload";
     {
@@ -2380,7 +2358,6 @@ Y_UNIT_TEST(MultiDestinationSinglePublication) {
     auto topicStub = MakeTopicServiceStub(server);
     const ui64 intPublicationId = BeginPublicationIntId(
         CallBeginPublication(*deferredStub, "/Root", "ext-lifecycle-multi-dest"));
-    GrantPublicationRegistryDelete(server, "root@builtin");
 
     constexpr TStringBuf payloadA = "lifecycle-topic-a";
     constexpr TStringBuf payloadB = "lifecycle-topic-b";
@@ -2409,7 +2386,6 @@ Y_UNIT_TEST(MultiDestinationSinglePublication) {
 
 Y_UNIT_TEST(RepeatFinalizeNotFound) {
     auto fixture = TDeferredStreamWriteFixture::Enabled("lifecycle-repeat-topic", "ext-lifecycle-repeat");
-    GrantPublicationRegistryDelete(fixture.Server, "root@builtin");
 
     constexpr TStringBuf payload = "lifecycle-repeat-payload";
     {
@@ -2449,7 +2425,6 @@ Y_UNIT_TEST(BeginOnlyPublishAborts) {
 
     const ui64 intPublicationId = BeginPublicationIntId(
         CallBeginPublication(*deferredStub, "/Root", "ext-begin-only-publish"));
-    GrantPublicationRegistryDelete(server, "root@builtin");
 
     const auto publishOutcome = CallPublish(*deferredStub, "/Root", intPublicationId);
     UNIT_ASSERT_VALUES_EQUAL(publishOutcome.Operation.status(), Ydb::StatusIds::ABORTED);
@@ -2467,7 +2442,6 @@ Y_UNIT_TEST(BeginOnlyCancelDeletes) {
 
     const ui64 intPublicationId = BeginPublicationIntId(
         CallBeginPublication(*deferredStub, "/Root", "ext-begin-only-cancel"));
-    GrantPublicationRegistryDelete(server, "root@builtin");
 
     const auto cancelOutcome = CallCancelPublication(*deferredStub, "/Root", intPublicationId);
     UNIT_ASSERT_VALUES_EQUAL(cancelOutcome.Operation.status(), Ydb::StatusIds::SUCCESS);