From b628119ed344e8618bd6d4a45f4d939606896c7a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 10 May 2026 13:07:18 +0000 Subject: [PATCH 1/5] DescribeNode: optional semaphore name listing via kesus GetConfig Add include_semaphore_names on DescribeNodeRequest and semaphore_names on DescribeNodeResult, populated by querying the kesus tablet (extended TEvGetConfig) only when requested. C++ SDK exposes IncludeSemaphoreNames on TDescribeNodeSettings and GetSemaphoreNames() on TNodeDescription. Tests cover tablet GetConfig and end-to-end DescribeNode behavior. Co-authored-by: Maksim Zinal --- .../rpc_describe_coordination_node.cpp | 86 ++++++++++++++++++- ydb/core/kesus/tablet/tablet_ut.cpp | 31 +++++++ ydb/core/kesus/tablet/tx_config_get.cpp | 18 +++- ydb/core/kesus/tablet/ut_helpers.cpp | 6 +- ydb/core/kesus/tablet/ut_helpers.h | 2 +- ydb/core/protos/kesus.proto | 3 +- ydb/public/api/protos/ydb_coordination.proto | 4 + .../client/coordination/coordination.h | 5 +- .../src/client/coordination/coordination.cpp | 7 ++ ydb/services/ydb/ydb_coordination_ut.cpp | 34 ++++++++ 10 files changed, 187 insertions(+), 9 deletions(-) diff --git a/ydb/core/grpc_services/rpc_describe_coordination_node.cpp b/ydb/core/grpc_services/rpc_describe_coordination_node.cpp index 4607f8dcef97f..0cfcd0adb0184 100644 --- a/ydb/core/grpc_services/rpc_describe_coordination_node.cpp +++ b/ydb/core/grpc_services/rpc_describe_coordination_node.cpp @@ -6,6 +6,8 @@ #include "rpc_scheme_base.h" #include "rpc_common/rpc_common.h" +#include +#include #include #include #include @@ -14,6 +16,7 @@ namespace NKikimr { namespace NGRpcService { using namespace NActors; +using namespace NKesus; using namespace Ydb; using TEvDescribeCoordinationNode = TGrpcRequestOperationCall& ev) { + if (WaitingForSemaphoreNames_) { + switch (ev->GetTypeRewrite()) { + HFunc(TEvTabletPipe::TEvClientConnected, HandleSemaphorePipeConnected); + HFunc(TEvTabletPipe::TEvClientDestroyed, HandleSemaphorePipeDestroyed); + HFunc(TEvKesus::TEvGetConfigResult, HandleGetConfigResult); + default: + TBase::StateWork(ev); + } + return; + } + switch (ev->GetTypeRewrite()) { HFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle); - default: TBase::StateWork(ev); + default: + TBase::StateWork(ev); + } + } + + void HandleSemaphorePipeConnected(TEvTabletPipe::TEvClientConnected::TPtr& ev, const TActorContext& ctx) { + if (ev->Get()->Status != NKikimrProto::OK) { + NYql::TIssues issues; + issues.AddIssue(MakeIssue(NKikimrIssues::TIssuesIds::DEFAULT_ERROR, + TStringBuilder() << "Tablet not available, status: " << (ui32)ev->Get()->Status)); + return Reply(Ydb::StatusIds::UNAVAILABLE, issues, ctx); } + + auto req = MakeHolder(); + req->Record.SetIncludeSemaphoreNames(true); + NTabletPipe::SendData(SelfId(), KesusPipeClient_, req.Release(), 0); + } + + void HandleSemaphorePipeDestroyed(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActorContext& ctx) { + TBase::Handle(ev, ctx); + } + + void HandleGetConfigResult(TEvKesus::TEvGetConfigResult::TPtr& ev, const TActorContext& ctx) { + const auto& record = ev->Get()->Record; + for (const auto& name : record.GetSemaphoreNames()) { + PendingResult_.add_semaphore_names(name); + } + return ReplyWithResult(Ydb::StatusIds::SUCCESS, PendingResult_, ctx); } void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev, const TActorContext& ctx) { @@ -60,6 +122,21 @@ class TDescribeCoordinationNode : public TRpcSchemeRequestActorinclude_semaphore_names()) { + if (!pathDescription.HasKesus()) { + return Reply(Ydb::StatusIds::BAD_REQUEST, ctx); + } + const ui64 tabletId = pathDescription.GetKesus().GetKesusTabletId(); + if (!tabletId) { + return Reply(Ydb::StatusIds::BAD_REQUEST, ctx); + } + PendingResult_ = std::move(result); + KesusTabletId_ = tabletId; + StartFetchSemaphoreNames(); + return; + } + return ReplyWithResult(Ydb::StatusIds::SUCCESS, result, ctx); } case NKikimrScheme::StatusPathDoesNotExist: @@ -89,6 +166,11 @@ class TDescribeCoordinationNode : public TRpcSchemeRequestActor p, const IFacilityProvider& f) { diff --git a/ydb/core/kesus/tablet/tablet_ut.cpp b/ydb/core/kesus/tablet/tablet_ut.cpp index bb3ed2a32f439..6f06b4b245730 100644 --- a/ydb/core/kesus/tablet/tablet_ut.cpp +++ b/ydb/core/kesus/tablet/tablet_ut.cpp @@ -80,6 +80,37 @@ Y_UNIT_TEST_SUITE(TKesusTest) { ctx.SetConfig(12345, MakeConfig("/foo/bar/baz"), 41, Ydb::StatusIds::PRECONDITION_FAILED); } + Y_UNIT_TEST(TestGetConfigSemaphoreNames) { + TTestContext ctx; + ctx.Setup(); + auto proxy = ctx.Runtime->AllocateEdgeActor(); + ctx.MustRegisterProxy(proxy, 1); + ctx.MustAttachSession(proxy, 1, 0, 30000); + ctx.SendAcquireLock(111, proxy, 1, 1, "Lock1", LOCK_MODE_EXCLUSIVE); + ctx.ExpectAcquireLockResult(111, proxy, 1); + + UNIT_ASSERT_VALUES_EQUAL(ctx.GetConfig(false).GetSemaphoreNames().size(), 0u); + + ctx.CreateSemaphore("Alpha", 1); + ctx.CreateSemaphore("Beta", 1); + + UNIT_ASSERT_VALUES_EQUAL(ctx.GetConfig(false).GetSemaphoreNames().size(), 0u); + + const auto namesRecord = ctx.GetConfig(true); + UNIT_ASSERT_VALUES_EQUAL(namesRecord.GetSemaphoreNames().size(), 2u); + ui32 alphaCount = 0; + ui32 betaCount = 0; + for (const TString& name : namesRecord.GetSemaphoreNames()) { + if (name == "Alpha") { + ++alphaCount; + } else if (name == "Beta") { + ++betaCount; + } + } + UNIT_ASSERT_VALUES_EQUAL(alphaCount, 1u); + UNIT_ASSERT_VALUES_EQUAL(betaCount, 1u); + } + Y_UNIT_TEST(TestRegisterProxy) { TTestContext ctx; ctx.Setup(); diff --git a/ydb/core/kesus/tablet/tx_config_get.cpp b/ydb/core/kesus/tablet/tx_config_get.cpp index 1aff7e97a00e2..c20c108769df0 100644 --- a/ydb/core/kesus/tablet/tx_config_get.cpp +++ b/ydb/core/kesus/tablet/tx_config_get.cpp @@ -6,13 +6,15 @@ namespace NKesus { struct TKesusTablet::TTxConfigGet : public TTxBase { const TActorId Sender; const ui64 Cookie; + const bool IncludeSemaphoreNames; THolder Reply; - TTxConfigGet(TSelf* self, const TActorId& sender, ui64 cookie) + TTxConfigGet(TSelf* self, const TActorId& sender, ui64 cookie, bool includeSemaphoreNames) : TTxBase(self) , Sender(sender) , Cookie(cookie) + , IncludeSemaphoreNames(includeSemaphoreNames) {} TTxType GetTxType() const override { return TXTYPE_CONFIG_GET; } @@ -35,6 +37,17 @@ struct TKesusTablet::TTxConfigGet : public TTxBase { config->set_rate_limiter_counters_mode(Self->RateLimiterCountersMode); Reply->Record.SetVersion(Self->ConfigVersion); Reply->Record.SetPath(Self->KesusPath); + if (IncludeSemaphoreNames) { + TVector names; + names.reserve(Self->Semaphores.size()); + for (const auto& kv : Self->Semaphores) { + names.push_back(kv.second.Name); + } + Sort(names.begin(), names.end()); + for (const TString& name : names) { + Reply->Record.AddSemaphoreNames(name); + } + } return true; } @@ -48,7 +61,8 @@ struct TKesusTablet::TTxConfigGet : public TTxBase { }; void TKesusTablet::Handle(TEvKesus::TEvGetConfig::TPtr& ev) { - Execute(new TTxConfigGet(this, ev->Sender, ev->Cookie), TActivationContext::AsActorContext()); + Execute(new TTxConfigGet(this, ev->Sender, ev->Cookie, ev->Get()->Record.GetIncludeSemaphoreNames()), + TActivationContext::AsActorContext()); } } diff --git a/ydb/core/kesus/tablet/ut_helpers.cpp b/ydb/core/kesus/tablet/ut_helpers.cpp index 5c9af3ee5b7d9..62b74caf362d6 100644 --- a/ydb/core/kesus/tablet/ut_helpers.cpp +++ b/ydb/core/kesus/tablet/ut_helpers.cpp @@ -160,10 +160,12 @@ void TTestContext::SendFromProxy(const TActorId& proxy, ui64 generation, IEventB cookie); } -NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig() { +NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig(bool includeSemaphoreNames) { const ui64 cookie = RandomNumber(); const auto edge = Runtime->AllocateEdgeActor(); - SendFromEdge(edge, new TEvKesus::TEvGetConfig(), cookie); + THolder req = MakeHolder(); + req->Record.SetIncludeSemaphoreNames(includeSemaphoreNames); + SendFromEdge(edge, std::move(req), cookie); auto result = ExpectEdgeEvent(edge, cookie); UNIT_ASSERT_VALUES_EQUAL_C(result->Record.GetConfig().path(), result->Record.GetPath(), "Record: " << result->Record); diff --git a/ydb/core/kesus/tablet/ut_helpers.h b/ydb/core/kesus/tablet/ut_helpers.h index 52632d5c9a584..47cc499f5913e 100644 --- a/ydb/core/kesus/tablet/ut_helpers.h +++ b/ydb/core/kesus/tablet/ut_helpers.h @@ -75,7 +75,7 @@ struct TTestContext { void SendFromProxy(const TActorId& proxy, ui64 generation, IEventBase* payload, ui64 cookie = 0); // set/get config requests - NKikimrKesus::TEvGetConfigResult GetConfig(); + NKikimrKesus::TEvGetConfigResult GetConfig(bool includeSemaphoreNames = false); NKikimrKesus::TEvSetConfigResult SetConfig(ui64 txId, const Ydb::Coordination::Config& config, ui64 version, Ydb::StatusIds::StatusCode status = Ydb::StatusIds::SUCCESS); // Makes a dummy request using this proxy/generation pair diff --git a/ydb/core/protos/kesus.proto b/ydb/core/protos/kesus.proto index 2bf2fb000d37d..89f1bbae15ad5 100644 --- a/ydb/core/protos/kesus.proto +++ b/ydb/core/protos/kesus.proto @@ -37,13 +37,14 @@ message TEvSetConfigResult { } message TEvGetConfig { - // nothing + bool IncludeSemaphoreNames = 1; } message TEvGetConfigResult { Ydb.Coordination.Config Config = 1; uint64 Version = 2; string Path = 3; + repeated string SemaphoreNames = 4; } message TEvDescribeProxies { diff --git a/ydb/public/api/protos/ydb_coordination.proto b/ydb/public/api/protos/ydb_coordination.proto index 378045e78e83a..b5ef87a7d5289 100644 --- a/ydb/public/api/protos/ydb_coordination.proto +++ b/ydb/public/api/protos/ydb_coordination.proto @@ -484,6 +484,8 @@ message DropNodeResponse { message DescribeNodeRequest { string path = 1; Ydb.Operations.OperationParams operation_params = 2; + // When true, DescribeNodeResult includes semaphore_names from the coordination node. + bool include_semaphore_names = 3; } message DescribeNodeResponse { @@ -493,4 +495,6 @@ message DescribeNodeResponse { message DescribeNodeResult { Ydb.Scheme.Entry self = 1; Config config = 2; + // Names of semaphores stored in the coordination node (filled only when requested). + repeated string semaphore_names = 3; } diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h index bb03bc4e4c3e8..5b28d2760faf2 100644 --- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h +++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h @@ -107,6 +107,7 @@ class TNodeDescription { const std::string& GetOwner() const; const std::vector& GetEffectivePermissions() const; + const std::vector& GetSemaphoreNames() const; const Ydb::Coordination::DescribeNodeResult& GetProto() const; void SerializeTo(Ydb::Coordination::CreateNodeRequest& creationRequest) const; @@ -201,7 +202,9 @@ struct TAlterNodeSettings : public TNodeSettings { }; struct TDropNodeSettings : public TOperationRequestSettings { using TOperationRequestSettings::TOperationRequestSettings; }; -struct TDescribeNodeSettings : public TOperationRequestSettings { }; +struct TDescribeNodeSettings : public TOperationRequestSettings { + FLUENT_SETTING_DEFAULT(bool, IncludeSemaphoreNames, false); +}; //////////////////////////////////////////////////////////////////////////////// diff --git a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp index 71d3853bd6adc..878001334af78 100644 --- a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp +++ b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp @@ -61,6 +61,7 @@ struct TNodeDescription::TImpl { RateLimiterCountersMode_ = static_cast(config.rate_limiter_counters_mode()); Owner_ = desc.self().owner(); PermissionToSchemeEntry(desc.self().effective_permissions(), &EffectivePermissions_); + SemaphoreNames_.assign(desc.semaphore_names().begin(), desc.semaphore_names().end()); Proto_ = desc; } @@ -77,6 +78,7 @@ struct TNodeDescription::TImpl { ERateLimiterCountersMode RateLimiterCountersMode_; std::string Owner_; std::vector EffectivePermissions_; + std::vector SemaphoreNames_; Ydb::Coordination::DescribeNodeResult Proto_; }; @@ -113,6 +115,10 @@ const std::vector& TNodeDescription::GetEffectivePermissi return Impl_->EffectivePermissions_; } +const std::vector& TNodeDescription::GetSemaphoreNames() const { + return Impl_->SemaphoreNames_; +} + const Ydb::Coordination::DescribeNodeResult& TNodeDescription::GetProto() const { return Impl_->Proto_; } @@ -1980,6 +1986,7 @@ TAsyncDescribeNodeResult TClient::DescribeNode( { auto request = MakeOperationRequest(settings); request.set_path(TStringType{path}); + request.set_include_semaphore_names(settings.IncludeSemaphoreNames_); return Impl_->DescribeNode(std::move(request), settings); } diff --git a/ydb/services/ydb/ydb_coordination_ut.cpp b/ydb/services/ydb/ydb_coordination_ut.cpp index 8f9dd0e1643cb..8e38f89ec8a51 100644 --- a/ydb/services/ydb/ydb_coordination_ut.cpp +++ b/ydb/services/ydb/ydb_coordination_ut.cpp @@ -275,6 +275,40 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { EStatus::NOT_FOUND); } + Y_UNIT_TEST(DescribeNodeSemaphoreNames) { + TKikimrWithGrpcAndRootSchema server; + TClientContext context(server); + + ExpectSuccess(context.Client.CreateNode("/Root/node1")); + + auto session = ExpectSuccess(context.Client.StartSession("/Root/node1")); + ExpectSuccess(session.CreateSemaphore("SemA", 3)); + ExpectSuccess(session.CreateSemaphore("SemB", 2)); + + { + auto desc = ExpectSuccess(context.Client.DescribeNode("/Root/node1")); + UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames().size(), 0u); + } + { + auto desc = ExpectSuccess( + context.Client.DescribeNode( + "/Root/node1", + NYdb::NCoordination::TDescribeNodeSettings().IncludeSemaphoreNames(true))); + UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames().size(), 2u); + UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames()[0], "SemA"); + UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames()[1], "SemB"); + } + + ExpectSuccess(session.DeleteSemaphore("SemA")); + + auto descAfterDelete = ExpectSuccess( + context.Client.DescribeNode( + "/Root/node1", + NYdb::NCoordination::TDescribeNodeSettings().IncludeSemaphoreNames(true))); + UNIT_ASSERT_VALUES_EQUAL(descAfterDelete.GetSemaphoreNames().size(), 1u); + UNIT_ASSERT_VALUES_EQUAL(descAfterDelete.GetSemaphoreNames()[0], "SemB"); + } + Y_UNIT_TEST(SessionMethods) { TKikimrWithGrpcAndRootSchema server; From a7f9464b174df4974727c78521c0711069f246b8 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 10 May 2026 14:01:09 +0000 Subject: [PATCH 2/5] Add ListSemaphores server stream; drop DescribeNode semaphore extension Introduce CoordinationService.ListSemaphores streaming SemaphoreDescription with include_details controlling compact vs full fields. Validate SelectRow like DescribeSemaphore via resolve proxy + security object. Kesus tablet handles TEvListSemaphores; proxy forwards as a direct request. C++ client exposes ListSemaphores() with TListSemaphoresIterator. Reverts DescribeNode optional semaphore names and GetConfig semaphore listing. Co-authored-by: Maksim Zinal --- .../rpc_describe_coordination_node.cpp | 86 +-------- ydb/core/kesus/proxy/proxy_actor.cpp | 12 ++ ydb/core/kesus/tablet/events.h | 27 +++ ydb/core/kesus/tablet/tablet_impl.cpp | 1 + ydb/core/kesus/tablet/tablet_impl.h | 3 + ydb/core/kesus/tablet/tablet_ut.cpp | 40 ++-- ydb/core/kesus/tablet/tx_config_get.cpp | 18 +- ydb/core/kesus/tablet/tx_semaphore_list.cpp | 100 ++++++++++ ydb/core/kesus/tablet/ut_helpers.cpp | 15 +- ydb/core/kesus/tablet/ut_helpers.h | 3 +- ydb/core/kesus/tablet/ya.make | 1 + ydb/core/protos/counters_kesus.proto | 1 + ydb/core/protos/kesus.proto | 15 +- ydb/public/api/grpc/ydb_coordination_v1.proto | 3 + ydb/public/api/protos/ydb_coordination.proto | 14 +- .../client/coordination/coordination.h | 56 +++++- .../src/client/coordination/coordination.cpp | 108 ++++++++++- ydb/services/kesus/grpc_list_semaphores.cpp | 172 ++++++++++++++++++ ydb/services/kesus/grpc_list_semaphores.h | 13 ++ ydb/services/kesus/grpc_service.cpp | 27 +++ ydb/services/kesus/ya.make | 2 + ydb/services/ydb/ydb_coordination_ut.cpp | 59 ++++-- 22 files changed, 620 insertions(+), 156 deletions(-) create mode 100644 ydb/core/kesus/tablet/tx_semaphore_list.cpp create mode 100644 ydb/services/kesus/grpc_list_semaphores.cpp create mode 100644 ydb/services/kesus/grpc_list_semaphores.h diff --git a/ydb/core/grpc_services/rpc_describe_coordination_node.cpp b/ydb/core/grpc_services/rpc_describe_coordination_node.cpp index 0cfcd0adb0184..4607f8dcef97f 100644 --- a/ydb/core/grpc_services/rpc_describe_coordination_node.cpp +++ b/ydb/core/grpc_services/rpc_describe_coordination_node.cpp @@ -6,8 +6,6 @@ #include "rpc_scheme_base.h" #include "rpc_common/rpc_common.h" -#include -#include #include #include #include @@ -16,7 +14,6 @@ namespace NKikimr { namespace NGRpcService { using namespace NActors; -using namespace NKesus; using namespace Ydb; using TEvDescribeCoordinationNode = TGrpcRequestOperationCall& ev) { - if (WaitingForSemaphoreNames_) { - switch (ev->GetTypeRewrite()) { - HFunc(TEvTabletPipe::TEvClientConnected, HandleSemaphorePipeConnected); - HFunc(TEvTabletPipe::TEvClientDestroyed, HandleSemaphorePipeDestroyed); - HFunc(TEvKesus::TEvGetConfigResult, HandleGetConfigResult); - default: - TBase::StateWork(ev); - } - return; - } - switch (ev->GetTypeRewrite()) { HFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle); - default: - TBase::StateWork(ev); - } - } - - void HandleSemaphorePipeConnected(TEvTabletPipe::TEvClientConnected::TPtr& ev, const TActorContext& ctx) { - if (ev->Get()->Status != NKikimrProto::OK) { - NYql::TIssues issues; - issues.AddIssue(MakeIssue(NKikimrIssues::TIssuesIds::DEFAULT_ERROR, - TStringBuilder() << "Tablet not available, status: " << (ui32)ev->Get()->Status)); - return Reply(Ydb::StatusIds::UNAVAILABLE, issues, ctx); + default: TBase::StateWork(ev); } - - auto req = MakeHolder(); - req->Record.SetIncludeSemaphoreNames(true); - NTabletPipe::SendData(SelfId(), KesusPipeClient_, req.Release(), 0); - } - - void HandleSemaphorePipeDestroyed(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActorContext& ctx) { - TBase::Handle(ev, ctx); - } - - void HandleGetConfigResult(TEvKesus::TEvGetConfigResult::TPtr& ev, const TActorContext& ctx) { - const auto& record = ev->Get()->Record; - for (const auto& name : record.GetSemaphoreNames()) { - PendingResult_.add_semaphore_names(name); - } - return ReplyWithResult(Ydb::StatusIds::SUCCESS, PendingResult_, ctx); } void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev, const TActorContext& ctx) { @@ -122,21 +60,6 @@ class TDescribeCoordinationNode : public TRpcSchemeRequestActorinclude_semaphore_names()) { - if (!pathDescription.HasKesus()) { - return Reply(Ydb::StatusIds::BAD_REQUEST, ctx); - } - const ui64 tabletId = pathDescription.GetKesus().GetKesusTabletId(); - if (!tabletId) { - return Reply(Ydb::StatusIds::BAD_REQUEST, ctx); - } - PendingResult_ = std::move(result); - KesusTabletId_ = tabletId; - StartFetchSemaphoreNames(); - return; - } - return ReplyWithResult(Ydb::StatusIds::SUCCESS, result, ctx); } case NKikimrScheme::StatusPathDoesNotExist: @@ -166,11 +89,6 @@ class TDescribeCoordinationNode : public TRpcSchemeRequestActor p, const IFacilityProvider& f) { diff --git a/ydb/core/kesus/proxy/proxy_actor.cpp b/ydb/core/kesus/proxy/proxy_actor.cpp index 7efc77244e0ed..ebf1901c062a1 100644 --- a/ydb/core/kesus/proxy/proxy_actor.cpp +++ b/ydb/core/kesus/proxy/proxy_actor.cpp @@ -530,6 +530,16 @@ class TKesusProxyActor : public TActorBootstrapped { } } + void Handle(TEvKesus::TEvListSemaphores::TPtr& ev) { + auto msg = ev->Release(); + HandleDirectRequest(ev->Sender, ev->Cookie, std::move(msg)); + } + + void Handle(TEvKesus::TEvListSemaphoresResult::TPtr& ev) { + auto msg = ev->Release(); + HandleDirectResponse(ev->Cookie, std::move(msg)); + } + void Handle(TEvKesus::TEvAttachSession::TPtr& ev) { KPROXY_LOG_TRACE_S("Received TEvAttachSession from " << ev->Sender); Y_ABORT_UNLESS(ev->Sender); @@ -823,6 +833,8 @@ class TKesusProxyActor : public TActorBootstrapped { hFunc(TEvKesus::TEvUpdateSemaphoreResult, Handle); hFunc(TEvKesus::TEvDeleteSemaphore, Handle); hFunc(TEvKesus::TEvDeleteSemaphoreResult, Handle); + hFunc(TEvKesus::TEvListSemaphores, Handle); + hFunc(TEvKesus::TEvListSemaphoresResult, Handle); hFunc(TEvKesus::TEvAttachSession, Handle); hFunc(TEvKesus::TEvAttachSessionResult, Handle); hFunc(TEvKesus::TEvProxyExpired, Handle); diff --git a/ydb/core/kesus/tablet/events.h b/ydb/core/kesus/tablet/events.h index 8b5b2de401076..07c5dde88fd25 100644 --- a/ydb/core/kesus/tablet/events.h +++ b/ydb/core/kesus/tablet/events.h @@ -58,6 +58,9 @@ namespace TEvKesus { EvReleaseSemaphore, EvReleaseSemaphoreResult, + EvListSemaphores, + EvListSemaphoresResult, + // Notifications EvProxyExpired = EvBegin + 512, EvSessionExpired, @@ -249,6 +252,30 @@ namespace TEvKesus { using TResultBase::TResultBase; }; + struct TEvListSemaphores : public TEventPB { + TEvListSemaphores() = default; + + TEvListSemaphores(const TString& kesusPath, bool includeDetails) { + Record.SetKesusPath(kesusPath); + Record.SetProxyGeneration(0); + Record.SetIncludeDetails(includeDetails); + } + }; + + struct TEvListSemaphoresResult : public TEventPB { + TEvListSemaphoresResult() = default; + + explicit TEvListSemaphoresResult(ui64 generation) { + Record.SetProxyGeneration(generation); + Record.MutableError()->SetStatus(Ydb::StatusIds::SUCCESS); + } + + TEvListSemaphoresResult(ui64 generation, Ydb::StatusIds::StatusCode status, const TString& reason) { + Record.SetProxyGeneration(generation); + FillError(Record.MutableError(), status, reason); + } + }; + struct TEvRegisterProxy : public TEventPB { TEvRegisterProxy() = default; diff --git a/ydb/core/kesus/tablet/tablet_impl.cpp b/ydb/core/kesus/tablet/tablet_impl.cpp index e53aaaf24a3c9..16c1a2968e66d 100644 --- a/ydb/core/kesus/tablet/tablet_impl.cpp +++ b/ydb/core/kesus/tablet/tablet_impl.cpp @@ -279,6 +279,7 @@ STFUNC(TKesusTablet::StateWork) { hFunc(TEvKesus::TEvAcquireSemaphore, Handle); hFunc(TEvKesus::TEvCreateSemaphore, Handle); hFunc(TEvKesus::TEvDescribeSemaphore, Handle); + hFunc(TEvKesus::TEvListSemaphores, Handle); hFunc(TEvKesus::TEvDeleteSemaphore, Handle); hFunc(TEvKesus::TEvReleaseSemaphore, Handle); hFunc(TEvKesus::TEvUpdateSemaphore, Handle); diff --git a/ydb/core/kesus/tablet/tablet_impl.h b/ydb/core/kesus/tablet/tablet_impl.h index 6d40117715d8d..68fa765f55f5b 100644 --- a/ydb/core/kesus/tablet/tablet_impl.h +++ b/ydb/core/kesus/tablet/tablet_impl.h @@ -50,6 +50,8 @@ class TKesusTablet : public TActor, public NTabletFlatExecutor::TT struct TTxSemaphoreTimeout; struct TTxSemaphoreUpdate; + struct TTxSemaphoreList; + struct TTxQuoterResourceAdd; struct TTxQuoterResourceUpdate; struct TTxQuoterResourceDelete; @@ -393,6 +395,7 @@ class TKesusTablet : public TActor, public NTabletFlatExecutor::TT void Handle(TEvKesus::TEvSetConfig::TPtr& ev); void Handle(TEvKesus::TEvGetConfig::TPtr& ev); void Handle(TEvKesus::TEvDescribeSemaphore::TPtr& ev); + void Handle(TEvKesus::TEvListSemaphores::TPtr& ev); void Handle(TEvKesus::TEvDescribeProxies::TPtr& ev); void Handle(TEvKesus::TEvDescribeSessions::TPtr& ev); void Handle(TEvKesus::TEvRegisterProxy::TPtr& ev); diff --git a/ydb/core/kesus/tablet/tablet_ut.cpp b/ydb/core/kesus/tablet/tablet_ut.cpp index 6f06b4b245730..557401490e3cf 100644 --- a/ydb/core/kesus/tablet/tablet_ut.cpp +++ b/ydb/core/kesus/tablet/tablet_ut.cpp @@ -80,7 +80,7 @@ Y_UNIT_TEST_SUITE(TKesusTest) { ctx.SetConfig(12345, MakeConfig("/foo/bar/baz"), 41, Ydb::StatusIds::PRECONDITION_FAILED); } - Y_UNIT_TEST(TestGetConfigSemaphoreNames) { + Y_UNIT_TEST(TestListSemaphoresFromTablet) { TTestContext ctx; ctx.Setup(); auto proxy = ctx.Runtime->AllocateEdgeActor(); @@ -89,26 +89,36 @@ Y_UNIT_TEST_SUITE(TKesusTest) { ctx.SendAcquireLock(111, proxy, 1, 1, "Lock1", LOCK_MODE_EXCLUSIVE); ctx.ExpectAcquireLockResult(111, proxy, 1); - UNIT_ASSERT_VALUES_EQUAL(ctx.GetConfig(false).GetSemaphoreNames().size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(ctx.ListSemaphoresFromTablet(false).GetSemaphoreDescriptions().size(), 0u); ctx.CreateSemaphore("Alpha", 1); ctx.CreateSemaphore("Beta", 1); - UNIT_ASSERT_VALUES_EQUAL(ctx.GetConfig(false).GetSemaphoreNames().size(), 0u); - - const auto namesRecord = ctx.GetConfig(true); - UNIT_ASSERT_VALUES_EQUAL(namesRecord.GetSemaphoreNames().size(), 2u); - ui32 alphaCount = 0; - ui32 betaCount = 0; - for (const TString& name : namesRecord.GetSemaphoreNames()) { - if (name == "Alpha") { - ++alphaCount; - } else if (name == "Beta") { - ++betaCount; + { + const auto r = ctx.ListSemaphoresFromTablet(false); + UNIT_ASSERT_VALUES_EQUAL(r.GetSemaphoreDescriptions().size(), 2u); + ui32 alphaCount = 0; + ui32 betaCount = 0; + for (const auto& d : r.GetSemaphoreDescriptions()) { + UNIT_ASSERT_VALUES_EQUAL(d.owners_size(), 0); + UNIT_ASSERT_VALUES_EQUAL(d.waiters_size(), 0); + if (d.name() == "Alpha") { + ++alphaCount; + } else if (d.name() == "Beta") { + ++betaCount; + } + } + UNIT_ASSERT_VALUES_EQUAL(alphaCount, 1u); + UNIT_ASSERT_VALUES_EQUAL(betaCount, 1u); + } + + { + const auto r = ctx.ListSemaphoresFromTablet(true); + UNIT_ASSERT_VALUES_EQUAL(r.GetSemaphoreDescriptions().size(), 2u); + for (const auto& d : r.GetSemaphoreDescriptions()) { + UNIT_ASSERT_VALUES_EQUAL(d.owners_size(), 0u); } } - UNIT_ASSERT_VALUES_EQUAL(alphaCount, 1u); - UNIT_ASSERT_VALUES_EQUAL(betaCount, 1u); } Y_UNIT_TEST(TestRegisterProxy) { diff --git a/ydb/core/kesus/tablet/tx_config_get.cpp b/ydb/core/kesus/tablet/tx_config_get.cpp index c20c108769df0..1aff7e97a00e2 100644 --- a/ydb/core/kesus/tablet/tx_config_get.cpp +++ b/ydb/core/kesus/tablet/tx_config_get.cpp @@ -6,15 +6,13 @@ namespace NKesus { struct TKesusTablet::TTxConfigGet : public TTxBase { const TActorId Sender; const ui64 Cookie; - const bool IncludeSemaphoreNames; THolder Reply; - TTxConfigGet(TSelf* self, const TActorId& sender, ui64 cookie, bool includeSemaphoreNames) + TTxConfigGet(TSelf* self, const TActorId& sender, ui64 cookie) : TTxBase(self) , Sender(sender) , Cookie(cookie) - , IncludeSemaphoreNames(includeSemaphoreNames) {} TTxType GetTxType() const override { return TXTYPE_CONFIG_GET; } @@ -37,17 +35,6 @@ struct TKesusTablet::TTxConfigGet : public TTxBase { config->set_rate_limiter_counters_mode(Self->RateLimiterCountersMode); Reply->Record.SetVersion(Self->ConfigVersion); Reply->Record.SetPath(Self->KesusPath); - if (IncludeSemaphoreNames) { - TVector names; - names.reserve(Self->Semaphores.size()); - for (const auto& kv : Self->Semaphores) { - names.push_back(kv.second.Name); - } - Sort(names.begin(), names.end()); - for (const TString& name : names) { - Reply->Record.AddSemaphoreNames(name); - } - } return true; } @@ -61,8 +48,7 @@ struct TKesusTablet::TTxConfigGet : public TTxBase { }; void TKesusTablet::Handle(TEvKesus::TEvGetConfig::TPtr& ev) { - Execute(new TTxConfigGet(this, ev->Sender, ev->Cookie, ev->Get()->Record.GetIncludeSemaphoreNames()), - TActivationContext::AsActorContext()); + Execute(new TTxConfigGet(this, ev->Sender, ev->Cookie), TActivationContext::AsActorContext()); } } diff --git a/ydb/core/kesus/tablet/tx_semaphore_list.cpp b/ydb/core/kesus/tablet/tx_semaphore_list.cpp new file mode 100644 index 0000000000000..8e9f10e7796e2 --- /dev/null +++ b/ydb/core/kesus/tablet/tx_semaphore_list.cpp @@ -0,0 +1,100 @@ +#include "tablet_impl.h" + +namespace NKikimr { +namespace NKesus { + +struct TKesusTablet::TTxSemaphoreList : public TTxBase { + const TActorId Sender; + const ui64 Cookie; + const NKikimrKesus::TEvListSemaphores Record; + + THolder Reply; + + TTxSemaphoreList(TSelf* self, const TActorId& sender, ui64 cookie, const NKikimrKesus::TEvListSemaphores& record) + : TTxBase(self) + , Sender(sender) + , Cookie(cookie) + , Record(record) + {} + + TTxType GetTxType() const override { return TXTYPE_SEMAPHORE_LIST; } + + bool Execute(TTransactionContext& txc, const TActorContext& ctx) override { + LOG_DEBUG_S(ctx, NKikimrServices::KESUS_TABLET, + "[" << Self->TabletID() << "] TTxSemaphoreList::Execute (sender=" << Sender + << ", cookie=" << Cookie << ")"); + + NIceDb::TNiceDb db(txc.DB); + + if (Record.GetProxyGeneration() != 0) { + Reply.Reset(new TEvKesus::TEvListSemaphoresResult( + Record.GetProxyGeneration(), + Ydb::StatusIds::BAD_REQUEST, + "Only direct proxy generation 0 is supported for listing semaphores")); + return true; + } + + if (Self->UseStrictRead()) { + Self->PersistStrictMarker(db); + } + + Reply.Reset(new TEvKesus::TEvListSemaphoresResult(0)); + + TVector sorted; + sorted.reserve(Self->Semaphores.size()); + for (auto& kv : Self->Semaphores) { + sorted.push_back(&kv.second); + } + Sort(sorted.begin(), sorted.end(), [](const TSemaphoreInfo* a, const TSemaphoreInfo* b) { + return a->Name < b->Name; + }); + + const bool includeDetails = Record.GetIncludeDetails(); + for (TSemaphoreInfo* semaphore : sorted) { + auto* desc = Reply->Record.AddSemaphoreDescriptions(); + desc->set_name(semaphore->Name); + desc->set_limit(semaphore->Limit); + desc->set_ephemeral(semaphore->Ephemeral); + desc->set_count(semaphore->Count); + if (includeDetails) { + desc->set_data(semaphore->Data); + for (const auto* owner : semaphore->Owners) { + auto* p = desc->add_owners(); + p->set_order_id(owner->OrderId); + p->set_session_id(owner->SessionId); + p->set_count(owner->Count); + p->set_data(owner->Data); + } + for (const auto& kv : semaphore->Waiters) { + auto* waiter = kv.second; + auto* p = desc->add_waiters(); + p->set_order_id(waiter->OrderId); + p->set_session_id(waiter->SessionId); + p->set_timeout_millis(waiter->TimeoutMillis); + p->set_count(waiter->Count); + p->set_data(waiter->Data); + } + } + } + + return true; + } + + void Complete(const TActorContext& ctx) override { + LOG_DEBUG_S(ctx, NKikimrServices::KESUS_TABLET, + "[" << Self->TabletID() << "] TTxSemaphoreList::Complete (sender=" << Sender + << ", cookie=" << Cookie << ")"); + Y_ABORT_UNLESS(Reply); + ctx.Send(Sender, Reply.Release(), 0, Cookie); + } +}; + +void TKesusTablet::Handle(TEvKesus::TEvListSemaphores::TPtr& ev) { + const auto& record = ev->Get()->Record; + VerifyKesusPath(record.GetKesusPath()); + + Execute(new TTxSemaphoreList(this, ev->Sender, ev->Cookie, record), TActivationContext::AsActorContext()); +} + +} +} diff --git a/ydb/core/kesus/tablet/ut_helpers.cpp b/ydb/core/kesus/tablet/ut_helpers.cpp index 62b74caf362d6..2ec3ffa9d7977 100644 --- a/ydb/core/kesus/tablet/ut_helpers.cpp +++ b/ydb/core/kesus/tablet/ut_helpers.cpp @@ -160,18 +160,25 @@ void TTestContext::SendFromProxy(const TActorId& proxy, ui64 generation, IEventB cookie); } -NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig(bool includeSemaphoreNames) { +NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig() { const ui64 cookie = RandomNumber(); const auto edge = Runtime->AllocateEdgeActor(); - THolder req = MakeHolder(); - req->Record.SetIncludeSemaphoreNames(includeSemaphoreNames); - SendFromEdge(edge, std::move(req), cookie); + SendFromEdge(edge, new TEvKesus::TEvGetConfig(), cookie); auto result = ExpectEdgeEvent(edge, cookie); UNIT_ASSERT_VALUES_EQUAL_C(result->Record.GetConfig().path(), result->Record.GetPath(), "Record: " << result->Record); return result->Record; } +NKikimrKesus::TEvListSemaphoresResult TTestContext::ListSemaphoresFromTablet(bool includeDetails) { + const ui64 cookie = RandomNumber(); + const auto edge = Runtime->AllocateEdgeActor(); + SendFromEdge(edge, new TEvKesus::TEvListSemaphores("", includeDetails), cookie); + + auto result = ExpectEdgeEvent(edge, cookie); + return result->Record; +} + NKikimrKesus::TEvSetConfigResult TTestContext::SetConfig(ui64 txId, const Ydb::Coordination::Config& config, ui64 version, Ydb::StatusIds::StatusCode status) { const ui64 cookie = RandomNumber(); const auto edge = Runtime->AllocateEdgeActor(); diff --git a/ydb/core/kesus/tablet/ut_helpers.h b/ydb/core/kesus/tablet/ut_helpers.h index 47cc499f5913e..b03771e8489f4 100644 --- a/ydb/core/kesus/tablet/ut_helpers.h +++ b/ydb/core/kesus/tablet/ut_helpers.h @@ -75,7 +75,8 @@ struct TTestContext { void SendFromProxy(const TActorId& proxy, ui64 generation, IEventBase* payload, ui64 cookie = 0); // set/get config requests - NKikimrKesus::TEvGetConfigResult GetConfig(bool includeSemaphoreNames = false); + NKikimrKesus::TEvGetConfigResult GetConfig(); + NKikimrKesus::TEvListSemaphoresResult ListSemaphoresFromTablet(bool includeDetails = false); NKikimrKesus::TEvSetConfigResult SetConfig(ui64 txId, const Ydb::Coordination::Config& config, ui64 version, Ydb::StatusIds::StatusCode status = Ydb::StatusIds::SUCCESS); // Makes a dummy request using this proxy/generation pair diff --git a/ydb/core/kesus/tablet/ya.make b/ydb/core/kesus/tablet/ya.make index ce4b14c6818a9..4be030f50c0ac 100644 --- a/ydb/core/kesus/tablet/ya.make +++ b/ydb/core/kesus/tablet/ya.make @@ -25,6 +25,7 @@ SRCS( tx_semaphore_create.cpp tx_semaphore_delete.cpp tx_semaphore_describe.cpp + tx_semaphore_list.cpp tx_semaphore_release.cpp tx_semaphore_timeout.cpp tx_semaphore_update.cpp diff --git a/ydb/core/protos/counters_kesus.proto b/ydb/core/protos/counters_kesus.proto index 33df5de6266fc..3b5590c16f9f9 100644 --- a/ydb/core/protos/counters_kesus.proto +++ b/ydb/core/protos/counters_kesus.proto @@ -75,4 +75,5 @@ enum ETxTypes { TXTYPE_QUOTER_RESOURCE_ADD = 19 [(TxTypeOpts) = {Name: "TxQouterResourceAdd"}]; TXTYPE_QUOTER_RESOURCE_UPDATE = 20 [(TxTypeOpts) = {Name: "TxQouterResourceUpdate"}]; TXTYPE_QUOTER_RESOURCE_DELETE = 21 [(TxTypeOpts) = {Name: "TxQouterResourceDelete"}]; + TXTYPE_SEMAPHORE_LIST = 22 [(TxTypeOpts) = {Name: "TxSemaphoreList"}]; } diff --git a/ydb/core/protos/kesus.proto b/ydb/core/protos/kesus.proto index 89f1bbae15ad5..425c2aea965d0 100644 --- a/ydb/core/protos/kesus.proto +++ b/ydb/core/protos/kesus.proto @@ -37,14 +37,13 @@ message TEvSetConfigResult { } message TEvGetConfig { - bool IncludeSemaphoreNames = 1; + // nothing } message TEvGetConfigResult { Ydb.Coordination.Config Config = 1; uint64 Version = 2; string Path = 3; - repeated string SemaphoreNames = 4; } message TEvDescribeProxies { @@ -140,6 +139,18 @@ message TEvDeleteSemaphoreResult { TKesusError Error = 2; } +message TEvListSemaphores { + string KesusPath = 1; + uint64 ProxyGeneration = 2; + bool IncludeDetails = 3; +} + +message TEvListSemaphoresResult { + uint64 ProxyGeneration = 1; + TKesusError Error = 2; + repeated Ydb.Coordination.SemaphoreDescription SemaphoreDescriptions = 3; +} + message TEvRegisterProxy { string KesusPath = 1; uint64 ProxyGeneration = 2; diff --git a/ydb/public/api/grpc/ydb_coordination_v1.proto b/ydb/public/api/grpc/ydb_coordination_v1.proto index c4120a2cd0240..10571722e2d40 100644 --- a/ydb/public/api/grpc/ydb_coordination_v1.proto +++ b/ydb/public/api/grpc/ydb_coordination_v1.proto @@ -31,4 +31,7 @@ service CoordinationService { // Describes a coordination node rpc DescribeNode(Coordination.DescribeNodeRequest) returns (Coordination.DescribeNodeResponse); + + // Lists semaphores in a coordination node (stream of SemaphoreDescription) + rpc ListSemaphores(Coordination.ListSemaphoresRequest) returns (stream Coordination.SemaphoreDescription); } diff --git a/ydb/public/api/protos/ydb_coordination.proto b/ydb/public/api/protos/ydb_coordination.proto index b5ef87a7d5289..f8bc4739aaa5a 100644 --- a/ydb/public/api/protos/ydb_coordination.proto +++ b/ydb/public/api/protos/ydb_coordination.proto @@ -484,8 +484,6 @@ message DropNodeResponse { message DescribeNodeRequest { string path = 1; Ydb.Operations.OperationParams operation_params = 2; - // When true, DescribeNodeResult includes semaphore_names from the coordination node. - bool include_semaphore_names = 3; } message DescribeNodeResponse { @@ -495,6 +493,14 @@ message DescribeNodeResponse { message DescribeNodeResult { Ydb.Scheme.Entry self = 1; Config config = 2; - // Names of semaphores stored in the coordination node (filled only when requested). - repeated string semaphore_names = 3; +} + +/** + * Lists semaphores stored in a coordination node (server-streaming RPC). + * With include_details = false only compact fields are filled (name, limit, ephemeral, count). + * With include_details = true the response matches DescribeSemaphore with owners and waiters included. + */ +message ListSemaphoresRequest { + string path = 1; + bool include_details = 2; } diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h index 5b28d2760faf2..d4bf084de3265 100644 --- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h +++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h @@ -1,12 +1,14 @@ #pragma once #include +#include namespace Ydb { namespace Coordination { class Config; class CreateNodeRequest; class DescribeNodeResult; + class ListSemaphoresRequest; class SemaphoreDescription; class SemaphoreSession; } @@ -107,7 +109,6 @@ class TNodeDescription { const std::string& GetOwner() const; const std::vector& GetEffectivePermissions() const; - const std::vector& GetSemaphoreNames() const; const Ydb::Coordination::DescribeNodeResult& GetProto() const; void SerializeTo(Ydb::Coordination::CreateNodeRequest& creationRequest) const; @@ -202,10 +203,42 @@ struct TAlterNodeSettings : public TNodeSettings { }; struct TDropNodeSettings : public TOperationRequestSettings { using TOperationRequestSettings::TOperationRequestSettings; }; -struct TDescribeNodeSettings : public TOperationRequestSettings { - FLUENT_SETTING_DEFAULT(bool, IncludeSemaphoreNames, false); +struct TDescribeNodeSettings : public TOperationRequestSettings { }; + +//////////////////////////////////////////////////////////////////////////////// + +struct TListSemaphoresSettings : public TOperationRequestSettings { + FLUENT_SETTING_DEFAULT(bool, IncludeDetails, false); +}; + +class TListSemaphoresPart : public TStreamPartStatus { +public: + bool HasSemaphoreDescription() const { + return Description_.has_value(); + } + + const TSemaphoreDescription& GetSemaphoreDescription() const { + return *Description_; + } + + explicit TListSemaphoresPart(TStatus&& status) + : TStreamPartStatus(std::move(status)) + {} + + TListSemaphoresPart(TStatus&& status, TSemaphoreDescription&& description) + : TStreamPartStatus(std::move(status)) + , Description_(std::move(description)) + {} + +private: + std::optional Description_; }; +using TAsyncListSemaphoresPart = NThreading::TFuture; + +class TListSemaphoresIterator; +using TAsyncListSemaphoresIterator = NThreading::TFuture; + //////////////////////////////////////////////////////////////////////////////// class TSession; @@ -315,11 +348,28 @@ class TClient { TAsyncDescribeNodeResult DescribeNode(const std::string& path, const TDescribeNodeSettings& settings = TDescribeNodeSettings()); + TAsyncListSemaphoresIterator ListSemaphores(const std::string& path, + const TListSemaphoresSettings& settings = TListSemaphoresSettings()); + private: class TImpl; std::shared_ptr Impl_; }; +class TListSemaphoresIterator : public TStatus { + friend class TClient::TImpl; + +public: + TAsyncListSemaphoresPart ReadNext(); + +private: + struct TReaderImpl; + + TListSemaphoresIterator(std::shared_ptr impl, TPlainStatus&& status); + + std::shared_ptr ReaderImpl_; +}; + //////////////////////////////////////////////////////////////////////////////// class TSessionContext; diff --git a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp index 878001334af78..12a52a0d88e19 100644 --- a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp +++ b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -61,7 +62,6 @@ struct TNodeDescription::TImpl { RateLimiterCountersMode_ = static_cast(config.rate_limiter_counters_mode()); Owner_ = desc.self().owner(); PermissionToSchemeEntry(desc.self().effective_permissions(), &EffectivePermissions_); - SemaphoreNames_.assign(desc.semaphore_names().begin(), desc.semaphore_names().end()); Proto_ = desc; } @@ -78,7 +78,6 @@ struct TNodeDescription::TImpl { ERateLimiterCountersMode RateLimiterCountersMode_; std::string Owner_; std::vector EffectivePermissions_; - std::vector SemaphoreNames_; Ydb::Coordination::DescribeNodeResult Proto_; }; @@ -115,10 +114,6 @@ const std::vector& TNodeDescription::GetEffectivePermissi return Impl_->EffectivePermissions_; } -const std::vector& TNodeDescription::GetSemaphoreNames() const { - return Impl_->SemaphoreNames_; -} - const Ydb::Coordination::DescribeNodeResult& TNodeDescription::GetProto() const { return Impl_->Proto_; } @@ -129,6 +124,67 @@ void TNodeDescription::SerializeTo(Ydb::Coordination::CreateNodeRequest& creatio //////////////////////////////////////////////////////////////////////////////// +struct TListSemaphoresIterator::TReaderImpl { +public: + using TStreamProcessorPtr = NYdbGrpc::IStreamRequestReadProcessor::TPtr; + using TGRpcStatus = NYdbGrpc::TGrpcStatus; + + TReaderImpl(TStreamProcessorPtr streamProcessor, std::string endpoint) + : StreamProcessor_(std::move(streamProcessor)) + , Endpoint_(std::move(endpoint)) + {} + + ~TReaderImpl() { + StreamProcessor_->Cancel(); + } + + bool IsFinished() const { + return Finished_; + } + + TAsyncListSemaphoresPart ReadNext(std::shared_ptr self) { + auto promise = NewPromise(); + auto readCb = [self, promise](TGRpcStatus&& grpcStatus) mutable { + if (!grpcStatus.Ok()) { + self->Finished_ = true; + if (grpcStatus.GRpcStatusCode == grpc::StatusCode::OUT_OF_RANGE) { + promise.SetValue(TListSemaphoresPart(TStatus(TPlainStatus( + EStatus::SUCCESS, NYdb::NIssue::TIssues{}, self->Endpoint_, {})))); + } else { + promise.SetValue(TListSemaphoresPart(TStatus(TPlainStatus( + std::move(grpcStatus), self->Endpoint_, {})))); + } + } else { + promise.SetValue(TListSemaphoresPart( + TStatus(TPlainStatus(EStatus::SUCCESS, NYdb::NIssue::TIssues{}, self->Endpoint_, {})), + TSemaphoreDescription(self->Response_))); + } + }; + StreamProcessor_->Read(&Response_, readCb); + return promise.GetFuture(); + } + +private: + TStreamProcessorPtr StreamProcessor_; + Ydb::Coordination::SemaphoreDescription Response_; + bool Finished_ = false; + std::string Endpoint_; +}; + +TListSemaphoresIterator::TListSemaphoresIterator(std::shared_ptr impl, TPlainStatus&& status) + : TStatus(std::move(status)) + , ReaderImpl_(std::move(impl)) +{} + +TAsyncListSemaphoresPart TListSemaphoresIterator::ReadNext() { + if (!ReaderImpl_ || ReaderImpl_->IsFinished()) { + RaiseError("Attempt to perform read on invalid or finished stream"); + } + return ReaderImpl_->ReadNext(ReaderImpl_); +} + +//////////////////////////////////////////////////////////////////////////////// + TSemaphoreSession::TSemaphoreSession() { OrderId_ = 0; SessionId_ = 0; @@ -1934,6 +1990,38 @@ class TClient::TImpl : public TClientImplCommon { return promise.GetFuture(); } + + TAsyncListSemaphoresIterator ListSemaphores( + const std::string& path, + const TListSemaphoresSettings& settings) + { + Ydb::Coordination::ListSemaphoresRequest request; + request.set_path(TStringType{path}); + request.set_include_details(settings.IncludeDetails_); + + auto promise = NewPromise(); + + Connections_->StartReadStream( + request, + [promise](TPlainStatus status, + NYdbGrpc::IStreamRequestReadProcessor::TPtr processor) mutable { + if (!status.Ok()) { + promise.SetValue(TListSemaphoresIterator(nullptr, std::move(status))); + return; + } + auto impl = std::make_shared( + std::move(processor), + status.Endpoint); + promise.SetValue(TListSemaphoresIterator(impl, std::move(status))); + }, + &Ydb::Coordination::V1::CoordinationService::Stub::AsyncListSemaphores, + DbDriverState_, + TRpcRequestSettings::Make(settings)); + + return promise.GetFuture(); + } }; TClient::TClient(const TDriver& driver, const TCommonClientSettings& settings) @@ -1986,10 +2074,16 @@ TAsyncDescribeNodeResult TClient::DescribeNode( { auto request = MakeOperationRequest(settings); request.set_path(TStringType{path}); - request.set_include_semaphore_names(settings.IncludeSemaphoreNames_); return Impl_->DescribeNode(std::move(request), settings); } +TAsyncListSemaphoresIterator TClient::ListSemaphores( + const std::string& path, + const TListSemaphoresSettings& settings) +{ + return Impl_->ListSemaphores(path, settings); +} + //////////////////////////////////////////////////////////////////////////////// class TSession::TImpl { diff --git a/ydb/services/kesus/grpc_list_semaphores.cpp b/ydb/services/kesus/grpc_list_semaphores.cpp new file mode 100644 index 0000000000000..470c984e37846 --- /dev/null +++ b/ydb/services/kesus/grpc_list_semaphores.cpp @@ -0,0 +1,172 @@ +#include "grpc_list_semaphores.h" + +#include +#include +#include + +#include +#include +#include +#include + +#include + +namespace NKikimr { +namespace NKesus { + +namespace { + +grpc::StatusCode YdbStatusToGrpcStatus(Ydb::StatusIds::StatusCode status) { + switch (status) { + case Ydb::StatusIds::NOT_FOUND: + return grpc::NOT_FOUND; + case Ydb::StatusIds::BAD_REQUEST: + return grpc::INVALID_ARGUMENT; + case Ydb::StatusIds::UNAUTHORIZED: + return grpc::PERMISSION_DENIED; + case Ydb::StatusIds::UNAVAILABLE: + return grpc::UNAVAILABLE; + default: + return grpc::INTERNAL; + } +} + +class TGRpcListSemaphoresActor : public TActorBootstrapped { +public: + explicit TGRpcListSemaphoresActor(TIntrusivePtr grpcRequest) + : GrpcRequest_(std::move(grpcRequest)) + {} + + static constexpr NKikimrServices::TActivity::EType ActorActivityType() { + return NKikimrServices::TActivity::KESUS_REQ; + } + + void Bootstrap(const TActorContext& ctx) { + Y_UNUSED(ctx); + + const auto* protoReq = dynamic_cast(GrpcRequest_->GetRequest()); + if (!protoReq || protoReq->path().empty()) { + GrpcRequest_->ReplyError(grpc::INVALID_ARGUMENT, "Invalid ListSemaphores request"); + return PassAway(); + } + + KesusPath_ = protoReq->path(); + IncludeDetails_ = protoReq->include_details(); + + auto dbVals = GrpcRequest_->GetPeerMetaValues(TStringBuf("x-ydb-database")); + if (!dbVals.empty()) { + Database_ = TString{dbVals[0]}; + } + + auto tokenVals = GrpcRequest_->GetPeerMetaValues(TStringBuf("x-ydb-auth-ticket")); + if (!tokenVals.empty() && !tokenVals[0].empty()) { + UserToken_.Reset(new NACLib::TUserToken(TString{tokenVals[0]})); + } + + if (!Send(MakeKesusProxyServiceId(), new TEvKesusProxy::TEvResolveKesusProxy(Database_, KesusPath_))) { + GrpcRequest_->ReplyError(grpc::UNIMPLEMENTED, "Coordination service not implemented on this server"); + return PassAway(); + } + + Become(&TThis::StateResolve); + } + +private: + bool CheckAccess(ui32 access) const { + if (UserToken_) { + if (!SecurityObject_) { + return true; + } + return SecurityObject_->CheckAccess(access, *UserToken_); + } + return true; + } + + void FailWithKesusError(const NKikimrKesus::TKesusError& err) { + TString msg = err.IssuesSize() ? TString{err.GetIssues(0).GetMessage()} : TString{"Coordination error"}; + GrpcRequest_->ReplyError(YdbStatusToGrpcStatus(err.GetStatus()), msg); + PassAway(); + } + + void HandleResolve(const TEvKesusProxy::TEvAttachProxyActor::TPtr& ev, const TActorContext& ctx) { + Y_UNUSED(ctx); + ProxyActor_ = ev->Get()->ProxyActor; + SecurityObject_ = ev->Get()->SecurityObject; + + const bool readAllowed = CheckAccess(NACLib::EAccessRights::SelectRow); + if (!readAllowed) { + GrpcRequest_->ReplyError(grpc::PERMISSION_DENIED, "Read permission denied"); + return PassAway(); + } + + ctx.Send(ProxyActor_, + new TEvKesus::TEvListSemaphores(KesusPath_, IncludeDetails_), + 0, + RequestCookie_); + + Become(&TThis::StateWaitTablet); + } + + void HandleProxyError(const TEvKesusProxy::TEvProxyError::TPtr& ev, const TActorContext& ctx) { + Y_UNUSED(ctx); + FailWithKesusError(ev->Get()->Error); + } + + void HandleListResult(const TEvKesus::TEvListSemaphoresResult::TPtr& ev, const TActorContext& ctx) { + Y_UNUSED(ctx); + const auto& record = ev->Get()->Record; + if (record.GetError().GetStatus() != Ydb::StatusIds::SUCCESS) { + FailWithKesusError(record.GetError()); + return; + } + + const auto& items = record.GetSemaphoreDescriptions(); + for (const auto& src : items) { + auto* msg = google::protobuf::Arena::CreateMessage(GrpcRequest_->GetArena()); + msg->CopyFrom(src); + GrpcRequest_->Reply(msg, 0); + } + + GrpcRequest_->FinishStreamingOk(); + PassAway(); + } + + STFUNC(StateResolve) { + switch (ev->GetTypeRewrite()) { + hFunc(TEvKesusProxy::TEvProxyError, HandleProxyError); + hFunc(TEvKesusProxy::TEvAttachProxyActor, HandleResolve); + default: + Y_ABORT("Unexpected event 0x%x for TGRpcListSemaphoresActor::StateResolve", ev->GetTypeRewrite()); + } + } + + STFUNC(StateWaitTablet) { + switch (ev->GetTypeRewrite()) { + hFunc(TEvKesusProxy::TEvProxyError, HandleProxyError); + hFunc(TEvKesus::TEvListSemaphoresResult, HandleListResult); + default: + Y_ABORT("Unexpected event 0x%x for TGRpcListSemaphoresActor::StateWaitTablet", ev->GetTypeRewrite()); + } + } + +private: + TIntrusivePtr GrpcRequest_; + TString Database_; + TString KesusPath_; + bool IncludeDetails_ = false; + std::unique_ptr UserToken_; + TActorId ProxyActor_; + TIntrusivePtr SecurityObject_; + static constexpr ui64 RequestCookie_ = 1; +}; + +} // namespace + +void StartGRpcListSemaphores(NActors::TActorSystem* actorSystem, NYdbGrpc::IRequestContextBase* grpcRequest) { + Y_ABORT_UNLESS(actorSystem); + Y_ABORT_UNLESS(grpcRequest); + actorSystem->Register(new TGRpcListSemaphoresActor(TIntrusivePtr(grpcRequest))); +} + +} +} diff --git a/ydb/services/kesus/grpc_list_semaphores.h b/ydb/services/kesus/grpc_list_semaphores.h new file mode 100644 index 0000000000000..6676ba8e74d27 --- /dev/null +++ b/ydb/services/kesus/grpc_list_semaphores.h @@ -0,0 +1,13 @@ +#pragma once + +#include + +namespace NYdbGrpc { +class IRequestContextBase; +} + +namespace NKikimr::NKesus { + +void StartGRpcListSemaphores(NActors::TActorSystem* actorSystem, NYdbGrpc::IRequestContextBase* grpcRequest); + +} diff --git a/ydb/services/kesus/grpc_service.cpp b/ydb/services/kesus/grpc_service.cpp index 5eed4a2a57b9e..909d01b4a6ee9 100644 --- a/ydb/services/kesus/grpc_service.cpp +++ b/ydb/services/kesus/grpc_service.cpp @@ -1,4 +1,5 @@ #include "grpc_service.h" +#include "grpc_list_semaphores.h" #include #include @@ -646,6 +647,30 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { #error SETUP_KESUS_STREAM_METHOD macro already defined #endif +#ifdef SETUP_KESUS_LIST_SEMAPHORES_STREAM +#error SETUP_KESUS_LIST_SEMAPHORES_STREAM macro already defined +#endif + +#define SETUP_KESUS_LIST_SEMAPHORES_STREAM() \ + for (auto* cq : CQS) { \ + MakeIntrusive<::NKikimr::NGRpcService::TGRpcRequest< \ + Ydb::Coordination::ListSemaphoresRequest, \ + Ydb::Coordination::SemaphoreDescription, \ + TKesusGRpcService>>( \ + this, \ + &Service_, \ + cq, \ + [this](NYdbGrpc::IRequestContextBase* reqCtx) { \ + StartGRpcListSemaphores(ActorSystem_, reqCtx); \ + }, \ + &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestListSemaphores, \ + "ListSemaphores", \ + logger, \ + getCounterBlock("coordination", "ListSemaphores", true), \ + nullptr \ + )->Run(); \ + } + #define SETUP_KESUS_METHOD(methodName, methodCallback, rlMode, requestType, auditMode) \ for (auto* cq : CQS) { \ SETUP_RUNTIME_EVENT_METHOD(methodName, \ @@ -689,11 +714,13 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { SETUP_KESUS_METHOD(AlterNode, DoAlterCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::Modifying(TAuditMode::TLogClassConfig::Ddl)); SETUP_KESUS_METHOD(DropNode, DoDropCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::Modifying(TAuditMode::TLogClassConfig::Ddl)); SETUP_KESUS_METHOD(DescribeNode, DoDescribeCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::NonModifying()); + SETUP_KESUS_LIST_SEMAPHORES_STREAM(); SETUP_KESUS_STREAM_METHOD(Session, RLMODE(Off), UNSPECIFIED, TAuditMode::NonModifying(), NGRpcService::TEvCoordinationSessionRequest); #undef GET_LIMITER_BY_PATH #undef SETUP_KESUS_METHOD #undef SETUP_KESUS_STREAM_METHOD +#undef SETUP_KESUS_LIST_SEMAPHORES_STREAM } } // namespace NKesus diff --git a/ydb/services/kesus/ya.make b/ydb/services/kesus/ya.make index 6a32f51a99a20..ecd7f6cdc58de 100644 --- a/ydb/services/kesus/ya.make +++ b/ydb/services/kesus/ya.make @@ -1,10 +1,12 @@ LIBRARY() SRCS( + grpc_list_semaphores.cpp grpc_service.cpp ) PEERDIR( + ydb/library/aclib ydb/library/grpc/server ydb/core/base ydb/core/grpc_services diff --git a/ydb/services/ydb/ydb_coordination_ut.cpp b/ydb/services/ydb/ydb_coordination_ut.cpp index 8e38f89ec8a51..f4c1b64504354 100644 --- a/ydb/services/ydb/ydb_coordination_ut.cpp +++ b/ydb/services/ydb/ydb_coordination_ut.cpp @@ -9,6 +9,8 @@ #include #include +#include + namespace NKikimr { using namespace Tests; @@ -275,7 +277,7 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { EStatus::NOT_FOUND); } - Y_UNIT_TEST(DescribeNodeSemaphoreNames) { + Y_UNIT_TEST(ListSemaphoresStream) { TKikimrWithGrpcAndRootSchema server; TClientContext context(server); @@ -286,30 +288,47 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { ExpectSuccess(session.CreateSemaphore("SemB", 2)); { - auto desc = ExpectSuccess(context.Client.DescribeNode("/Root/node1")); - UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames().size(), 0u); - } - { - auto desc = ExpectSuccess( - context.Client.DescribeNode( - "/Root/node1", - NYdb::NCoordination::TDescribeNodeSettings().IncludeSemaphoreNames(true))); - UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames().size(), 2u); - UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames()[0], "SemA"); - UNIT_ASSERT_VALUES_EQUAL(desc.GetSemaphoreNames()[1], "SemB"); + auto itFuture = context.Client.ListSemaphores( + "/Root/node1", + NYdb::NCoordination::TListSemaphoresSettings().IncludeDetails(false)); + auto it = itFuture.ExtractValueSync(); + UNIT_ASSERT_C(it.IsSuccess(), TStatusDescription(it)); + + std::unordered_set names; + while (true) { + auto part = it.ReadNext().ExtractValueSync(); + UNIT_ASSERT_C(part.IsSuccess(), TStatusDescription(part)); + if (!part.HasSemaphoreDescription()) { + break; + } + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetOwners().size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetWaiters().size(), 0u); + names.insert(part.GetSemaphoreDescription().GetName()); + } + UNIT_ASSERT(names.count("SemA")); + UNIT_ASSERT(names.count("SemB")); } - ExpectSuccess(session.DeleteSemaphore("SemA")); - - auto descAfterDelete = ExpectSuccess( - context.Client.DescribeNode( + { + auto itFuture = context.Client.ListSemaphores( "/Root/node1", - NYdb::NCoordination::TDescribeNodeSettings().IncludeSemaphoreNames(true))); - UNIT_ASSERT_VALUES_EQUAL(descAfterDelete.GetSemaphoreNames().size(), 1u); - UNIT_ASSERT_VALUES_EQUAL(descAfterDelete.GetSemaphoreNames()[0], "SemB"); + NYdb::NCoordination::TListSemaphoresSettings().IncludeDetails(true)); + auto it = itFuture.ExtractValueSync(); + UNIT_ASSERT_C(it.IsSuccess(), TStatusDescription(it)); + + size_t n = 0; + while (true) { + auto part = it.ReadNext().ExtractValueSync(); + UNIT_ASSERT_C(part.IsSuccess(), TStatusDescription(part)); + if (!part.HasSemaphoreDescription()) { + break; + } + ++n; + } + UNIT_ASSERT_VALUES_EQUAL(n, 2u); + } } - Y_UNIT_TEST(SessionMethods) { TKikimrWithGrpcAndRootSchema server; TClientContext context(server); From be46290c55bf7eec06ffd60e57e3b21ce655e5c4 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 10 May 2026 16:13:50 +0000 Subject: [PATCH 3/5] Address ListSemaphores review: aclib, operation_params, gRPC wiring, tests Co-authored-by: Maksim Zinal --- .../control/lib/generated/codegen/main.cpp | 1 + ydb/core/kesus/tablet/tablet_ut.cpp | 9 +++ ydb/public/api/protos/ydb_coordination.proto | 1 + .../client/coordination/coordination.h | 3 +- .../src/client/coordination/coordination.cpp | 2 +- ydb/services/kesus/grpc_list_semaphores.cpp | 23 +++--- ydb/services/kesus/grpc_service.cpp | 46 ++++++----- ydb/services/ydb/ydb_coordination_ut.cpp | 78 +++++++++++++++++-- 8 files changed, 122 insertions(+), 41 deletions(-) diff --git a/ydb/core/control/lib/generated/codegen/main.cpp b/ydb/core/control/lib/generated/codegen/main.cpp index b6403aab18d18..bca55ed96b8be 100644 --- a/ydb/core/control/lib/generated/codegen/main.cpp +++ b/ydb/core/control/lib/generated/codegen/main.cpp @@ -172,6 +172,7 @@ void CodeGenRequestConfigsInner(TCodeGenContext& context); std::vector GetRequestConfigsServices() { return { "CoordinationService_Session", + "CoordinationService_ListSemaphores", "ClickhouseInternal_Scan", "ClickhouseInternal_GetShardLocations", "ClickhouseInternal_DescribeTable", diff --git a/ydb/core/kesus/tablet/tablet_ut.cpp b/ydb/core/kesus/tablet/tablet_ut.cpp index 557401490e3cf..757cbf974c57e 100644 --- a/ydb/core/kesus/tablet/tablet_ut.cpp +++ b/ydb/core/kesus/tablet/tablet_ut.cpp @@ -93,6 +93,7 @@ Y_UNIT_TEST_SUITE(TKesusTest) { ctx.CreateSemaphore("Alpha", 1); ctx.CreateSemaphore("Beta", 1); + ctx.UpdateSemaphore("Alpha", "meta-alpha"); { const auto r = ctx.ListSemaphoresFromTablet(false); @@ -117,6 +118,14 @@ Y_UNIT_TEST_SUITE(TKesusTest) { UNIT_ASSERT_VALUES_EQUAL(r.GetSemaphoreDescriptions().size(), 2u); for (const auto& d : r.GetSemaphoreDescriptions()) { UNIT_ASSERT_VALUES_EQUAL(d.owners_size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(d.waiters_size(), 0u); + if (d.name() == "Alpha") { + UNIT_ASSERT_VALUES_EQUAL(d.data(), "meta-alpha"); + } else if (d.name() == "Beta") { + UNIT_ASSERT_VALUES_EQUAL(d.data(), ""); + } else { + UNIT_FAIL("unexpected semaphore"); + } } } } diff --git a/ydb/public/api/protos/ydb_coordination.proto b/ydb/public/api/protos/ydb_coordination.proto index f8bc4739aaa5a..1b41268efe5ed 100644 --- a/ydb/public/api/protos/ydb_coordination.proto +++ b/ydb/public/api/protos/ydb_coordination.proto @@ -503,4 +503,5 @@ message DescribeNodeResult { message ListSemaphoresRequest { string path = 1; bool include_details = 2; + Ydb.Operations.OperationParams operation_params = 3; } diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h index d4bf084de3265..847d2f407fe5c 100644 --- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h +++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h @@ -329,6 +329,8 @@ struct TDescribeSemaphoreSettings { class TClient { public: + class TImpl; + TClient(const TDriver& driver, const TCommonClientSettings& settings = TCommonClientSettings()); ~TClient(); @@ -352,7 +354,6 @@ class TClient { const TListSemaphoresSettings& settings = TListSemaphoresSettings()); private: - class TImpl; std::shared_ptr Impl_; }; diff --git a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp index 12a52a0d88e19..a13ebf6885061 100644 --- a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp +++ b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp @@ -1995,7 +1995,7 @@ class TClient::TImpl : public TClientImplCommon { const std::string& path, const TListSemaphoresSettings& settings) { - Ydb::Coordination::ListSemaphoresRequest request; + auto request = MakeOperationRequest(settings); request.set_path(TStringType{path}); request.set_include_details(settings.IncludeDetails_); diff --git a/ydb/services/kesus/grpc_list_semaphores.cpp b/ydb/services/kesus/grpc_list_semaphores.cpp index 470c984e37846..9b86859484640 100644 --- a/ydb/services/kesus/grpc_list_semaphores.cpp +++ b/ydb/services/kesus/grpc_list_semaphores.cpp @@ -4,13 +4,16 @@ #include #include -#include +#include #include +#include #include #include #include +#include + namespace NKikimr { namespace NKesus { @@ -37,10 +40,6 @@ class TGRpcListSemaphoresActor : public TActorBootstrappedGetPeerMetaValues(TStringBuf("x-ydb-auth-ticket")); if (!tokenVals.empty() && !tokenVals[0].empty()) { - UserToken_.Reset(new NACLib::TUserToken(TString{tokenVals[0]})); + UserToken_ = std::make_unique(TString{tokenVals[0]}); } if (!Send(MakeKesusProxyServiceId(), new TEvKesusProxy::TEvResolveKesusProxy(Database_, KesusPath_))) { @@ -83,13 +82,13 @@ class TGRpcListSemaphoresActor : public TActorBootstrappedReplyError(YdbStatusToGrpcStatus(err.GetStatus()), msg); PassAway(); } - void HandleResolve(const TEvKesusProxy::TEvAttachProxyActor::TPtr& ev, const TActorContext& ctx) { - Y_UNUSED(ctx); + void HandleResolve(const TEvKesusProxy::TEvAttachProxyActor::TPtr& ev) { + const TActorContext& ctx = ActorContext(); ProxyActor_ = ev->Get()->ProxyActor; SecurityObject_ = ev->Get()->SecurityObject; @@ -107,13 +106,11 @@ class TGRpcListSemaphoresActor : public TActorBootstrappedGet()->Error); } - void HandleListResult(const TEvKesus::TEvListSemaphoresResult::TPtr& ev, const TActorContext& ctx) { - Y_UNUSED(ctx); + void HandleListResult(const TEvKesus::TEvListSemaphoresResult::TPtr& ev) { const auto& record = ev->Get()->Record; if (record.GetError().GetStatus() != Ydb::StatusIds::SUCCESS) { FailWithKesusError(record.GetError()); diff --git a/ydb/services/kesus/grpc_service.cpp b/ydb/services/kesus/grpc_service.cpp index 909d01b4a6ee9..c261d25ad8d0e 100644 --- a/ydb/services/kesus/grpc_service.cpp +++ b/ydb/services/kesus/grpc_service.cpp @@ -651,26 +651,6 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { #error SETUP_KESUS_LIST_SEMAPHORES_STREAM macro already defined #endif -#define SETUP_KESUS_LIST_SEMAPHORES_STREAM() \ - for (auto* cq : CQS) { \ - MakeIntrusive<::NKikimr::NGRpcService::TGRpcRequest< \ - Ydb::Coordination::ListSemaphoresRequest, \ - Ydb::Coordination::SemaphoreDescription, \ - TKesusGRpcService>>( \ - this, \ - &Service_, \ - cq, \ - [this](NYdbGrpc::IRequestContextBase* reqCtx) { \ - StartGRpcListSemaphores(ActorSystem_, reqCtx); \ - }, \ - &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestListSemaphores, \ - "ListSemaphores", \ - logger, \ - getCounterBlock("coordination", "ListSemaphores", true), \ - nullptr \ - )->Run(); \ - } - #define SETUP_KESUS_METHOD(methodName, methodCallback, rlMode, requestType, auditMode) \ for (auto* cq : CQS) { \ SETUP_RUNTIME_EVENT_METHOD(methodName, \ @@ -693,6 +673,32 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { #define GET_LIMITER_BY_PATH(ICB_PATH) \ getLimiter(#ICB_PATH, icb.ICB_PATH, DEFAULT_MAX_SESSIONS_INFLIGHT) +// Server-streaming read (unary ListSemaphoresRequest, stream SemaphoreDescription). Uses TGRpcRequest's async-writer +// path like Session, but runs in-process on the Kesus GRpc service instead of TEvCoordinationSessionRequest. +#define SETUP_KESUS_LIST_SEMAPHORES_STREAM() \ + for (auto* cq : CQS) { \ + MakeIntrusive<::NKikimr::NGRpcService::TGRpcRequest< \ + Ydb::Coordination::ListSemaphoresRequest, \ + Ydb::Coordination::SemaphoreDescription, \ + TKesusGRpcService>>( \ + this, \ + &Service_, \ + cq, \ + [this](NYdbGrpc::IRequestContextBase* reqCtx) { \ + ::NKikimr::NGRpcService::ReportGrpcReqToMon( \ + *ActorSystem_, \ + reqCtx->GetPeer(), \ + GetSdkBuildInfoIfNeeded(reqCtx)); \ + StartGRpcListSemaphores(ActorSystem_, reqCtx); \ + }, \ + &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestListSemaphores, \ + "ListSemaphores", \ + logger, \ + getCounterBlock("coordination", "ListSemaphores", true), \ + GET_LIMITER_BY_PATH(GRpcControls.RequestConfigs.CoordinationService_ListSemaphores.MaxInFlight) \ + )->Run(); \ + } + #define SETUP_KESUS_STREAM_METHOD(methodName, rlMode, requestType, auditMode, operationCallClass) \ for (auto* cq : CQS) { \ SETUP_RUNTIME_EVENT_STREAM_METHOD(methodName, \ diff --git a/ydb/services/ydb/ydb_coordination_ut.cpp b/ydb/services/ydb/ydb_coordination_ut.cpp index f4c1b64504354..0eca95286a54e 100644 --- a/ydb/services/ydb/ydb_coordination_ut.cpp +++ b/ydb/services/ydb/ydb_coordination_ut.cpp @@ -9,6 +9,7 @@ #include #include +#include #include namespace NKikimr { @@ -66,6 +67,36 @@ struct TClientContext { } }; +void AssertSemaphoreSessionsEqual( + const NYdb::NCoordination::TSemaphoreSession& a, + const NYdb::NCoordination::TSemaphoreSession& b) +{ + UNIT_ASSERT_VALUES_EQUAL(a.GetOrderId(), b.GetOrderId()); + UNIT_ASSERT_VALUES_EQUAL(a.GetSessionId(), b.GetSessionId()); + UNIT_ASSERT_VALUES_EQUAL(a.GetCount(), b.GetCount()); + UNIT_ASSERT_VALUES_EQUAL(a.GetData(), b.GetData()); + UNIT_ASSERT_VALUES_EQUAL(a.GetTimeout(), b.GetTimeout()); +} + +void AssertSemaphoreDescriptionMatchesDescribe( + const NYdb::NCoordination::TSemaphoreDescription& listDesc, + const NYdb::NCoordination::TSemaphoreDescription& describeDesc) +{ + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetName(), describeDesc.GetName()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetData(), describeDesc.GetData()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetCount(), describeDesc.GetCount()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetLimit(), describeDesc.GetLimit()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.IsEphemeral(), describeDesc.IsEphemeral()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetOwners().size(), describeDesc.GetOwners().size()); + for (size_t i = 0; i < listDesc.GetOwners().size(); ++i) { + AssertSemaphoreSessionsEqual(listDesc.GetOwners()[i], describeDesc.GetOwners()[i]); + } + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetWaiters().size(), describeDesc.GetWaiters().size()); + for (size_t i = 0; i < listDesc.GetWaiters().size(); ++i) { + AssertSemaphoreSessionsEqual(listDesc.GetWaiters()[i], describeDesc.GetWaiters()[i]); + } +} + template class TSimpleQueue : public TThrRefBase { private: @@ -283,9 +314,36 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { ExpectSuccess(context.Client.CreateNode("/Root/node1")); - auto session = ExpectSuccess(context.Client.StartSession("/Root/node1")); - ExpectSuccess(session.CreateSemaphore("SemA", 3)); - ExpectSuccess(session.CreateSemaphore("SemB", 2)); + auto session1 = ExpectSuccess( + context.Client.StartSession("/Root/node1", + TSessionSettings().Timeout(TDuration::Seconds(30)))); + ExpectSuccess(session1.CreateSemaphore("SemA", 3)); + ExpectSuccess(session1.CreateSemaphore("SemB", 2)); + ExpectSuccess(session1.UpdateSemaphore("SemA", "meta-a")); + UNIT_ASSERT(ExpectSuccess( + session1.AcquireSemaphore("SemA", + TAcquireSemaphoreSettings() + .Count(2) + .Data("owner-a")))); + + auto session2 = ExpectSuccess( + context.Client.StartSession("/Root/node1", + TSessionSettings().Timeout(TDuration::Seconds(30)))); + auto waitAcquire = session2.AcquireSemaphore("SemA", + TAcquireSemaphoreSettings() + .Count(2) + .Timeout(TDuration::Max())); + + auto describeSemA = ExpectSuccess( + session1.DescribeSemaphore("SemA", + TDescribeSemaphoreSettings() + .IncludeOwners() + .IncludeWaiters())); + auto describeSemB = ExpectSuccess( + session1.DescribeSemaphore("SemB", + TDescribeSemaphoreSettings() + .IncludeOwners() + .IncludeWaiters())); { auto itFuture = context.Client.ListSemaphores( @@ -303,6 +361,7 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { } UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetOwners().size(), 0u); UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetWaiters().size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetData(), ""); names.insert(part.GetSemaphoreDescription().GetName()); } UNIT_ASSERT(names.count("SemA")); @@ -316,17 +375,24 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { auto it = itFuture.ExtractValueSync(); UNIT_ASSERT_C(it.IsSuccess(), TStatusDescription(it)); - size_t n = 0; + std::unordered_map byName; while (true) { auto part = it.ReadNext().ExtractValueSync(); UNIT_ASSERT_C(part.IsSuccess(), TStatusDescription(part)); if (!part.HasSemaphoreDescription()) { break; } - ++n; + const auto& d = part.GetSemaphoreDescription(); + byName[d.GetName()] = d; } - UNIT_ASSERT_VALUES_EQUAL(n, 2u); + UNIT_ASSERT_VALUES_EQUAL(byName.size(), 2u); + AssertSemaphoreDescriptionMatchesDescribe(byName.at("SemA"), describeSemA); + AssertSemaphoreDescriptionMatchesDescribe(byName.at("SemB"), describeSemB); } + + UNIT_ASSERT(ExpectSuccess(session1.ReleaseSemaphore("SemA"))); + UNIT_ASSERT(ExpectSuccess(std::move(waitAcquire))); + UNIT_ASSERT(ExpectSuccess(session2.ReleaseSemaphore("SemA"))); } Y_UNIT_TEST(SessionMethods) { From 2e26d4394cb3895407b88eeda6aa722e361ac5b9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 10 May 2026 16:49:28 +0000 Subject: [PATCH 4/5] test: wait OnAccepted before ListSemaphores waiter assertions Co-authored-by: Maksim Zinal --- ydb/services/ydb/ydb_coordination_ut.cpp | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/ydb/services/ydb/ydb_coordination_ut.cpp b/ydb/services/ydb/ydb_coordination_ut.cpp index 0eca95286a54e..af53b55658f04 100644 --- a/ydb/services/ydb/ydb_coordination_ut.cpp +++ b/ydb/services/ydb/ydb_coordination_ut.cpp @@ -329,10 +329,16 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { auto session2 = ExpectSuccess( context.Client.StartSession("/Root/node1", TSessionSettings().Timeout(TDuration::Seconds(30)))); + TPromise waiterAccepted = NewPromise(); + auto waiterAcceptedLambda = [=]() mutable { + waiterAccepted.SetValue(); + }; auto waitAcquire = session2.AcquireSemaphore("SemA", TAcquireSemaphoreSettings() .Count(2) - .Timeout(TDuration::Max())); + .Timeout(TDuration::Max()) + .OnAccepted(waiterAcceptedLambda)); + waiterAccepted.GetFuture().GetValueSync(); auto describeSemA = ExpectSuccess( session1.DescribeSemaphore("SemA", From 36bcae71035b22589d7d9fdb4a895d129ca7ea91 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 10 May 2026 18:29:30 +0000 Subject: [PATCH 5/5] test: drop lock setup from TestListSemaphoresFromTablet (lock maps to Semaphore) Co-authored-by: Maksim Zinal --- ydb/core/kesus/tablet/tablet_ut.cpp | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/ydb/core/kesus/tablet/tablet_ut.cpp b/ydb/core/kesus/tablet/tablet_ut.cpp index 757cbf974c57e..19727ceea2752 100644 --- a/ydb/core/kesus/tablet/tablet_ut.cpp +++ b/ydb/core/kesus/tablet/tablet_ut.cpp @@ -83,11 +83,9 @@ Y_UNIT_TEST_SUITE(TKesusTest) { Y_UNIT_TEST(TestListSemaphoresFromTablet) { TTestContext ctx; ctx.Setup(); - auto proxy = ctx.Runtime->AllocateEdgeActor(); - ctx.MustRegisterProxy(proxy, 1); - ctx.MustAttachSession(proxy, 1, 0, 30000); - ctx.SendAcquireLock(111, proxy, 1, 1, "Lock1", LOCK_MODE_EXCLUSIVE); - ctx.ExpectAcquireLockResult(111, proxy, 1); + + // Locks are acquired via TEvAcquireSemaphore and appear in Semaphores; keep this test focused on + // explicit CreateSemaphore entries only (no proxy/session lock setup needed for ListSemaphores). UNIT_ASSERT_VALUES_EQUAL(ctx.ListSemaphoresFromTablet(false).GetSemaphoreDescriptions().size(), 0u);