ListSemaphores streaming RPC and review fixes

ydb/core/control/lib/generated/codegen/main.cpp

@@ -172,6 +172,7 @@ void CodeGenRequestConfigsInner(TCodeGenContext& context);
 std::vector<TString> GetRequestConfigsServices() {
     return {
         "CoordinationService_Session",
+        "CoordinationService_ListSemaphores",
         "ClickhouseInternal_Scan",
         "ClickhouseInternal_GetShardLocations",
         "ClickhouseInternal_DescribeTable",

ydb/core/kesus/proxy/proxy_actor.cpp

@@ -530,6 +530,16 @@ class TKesusProxyActor : public TActorBootstrapped<TKesusProxyActor> {
         }
     }
 
+    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<TKesusProxyActor> {
             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);

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, NKikimrKesus::TEvListSemaphores, EvListSemaphores> {
+        TEvListSemaphores() = default;
+
+        TEvListSemaphores(const TString& kesusPath, bool includeDetails) {
+            Record.SetKesusPath(kesusPath);
+            Record.SetProxyGeneration(0);
+            Record.SetIncludeDetails(includeDetails);
+        }
+    };
+
+    struct TEvListSemaphoresResult : public TEventPB<TEvListSemaphoresResult, NKikimrKesus::TEvListSemaphoresResult, EvListSemaphoresResult> {
+        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, NKikimrKesus::TEvRegisterProxy, EvRegisterProxy> {
         TEvRegisterProxy() = default;
 

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);

ydb/core/kesus/tablet/tablet_impl.h

@@ -50,6 +50,8 @@ class TKesusTablet : public TActor<TKesusTablet>, public NTabletFlatExecutor::TT
     struct TTxSemaphoreTimeout;
     struct TTxSemaphoreUpdate;
 
+    struct TTxSemaphoreList;
+
     struct TTxQuoterResourceAdd;
     struct TTxQuoterResourceUpdate;
     struct TTxQuoterResourceDelete;
@@ -393,6 +395,7 @@ class TKesusTablet : public TActor<TKesusTablet>, 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);

ydb/core/kesus/tablet/tablet_ut.cpp

@@ -80,6 +80,54 @@ Y_UNIT_TEST_SUITE(TKesusTest) {
         ctx.SetConfig(12345, MakeConfig("/foo/bar/baz"), 41, Ydb::StatusIds::PRECONDITION_FAILED);
     }
 
+    Y_UNIT_TEST(TestListSemaphoresFromTablet) {
+        TTestContext ctx;
+        ctx.Setup();
+
+        // 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);
+
+        ctx.CreateSemaphore("Alpha", 1);
+        ctx.CreateSemaphore("Beta", 1);
+        ctx.UpdateSemaphore("Alpha", "meta-alpha");
+
+        {
+            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(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");
+                }
+            }
+        }
+    }
+
     Y_UNIT_TEST(TestRegisterProxy) {
         TTestContext ctx;
         ctx.Setup();

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<TEvKesus::TEvListSemaphoresResult> 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<TSemaphoreInfo*> 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());
+}
+
+}
+}

ydb/core/kesus/tablet/ut_helpers.cpp

@@ -170,6 +170,15 @@ NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig() {
     return result->Record;
 }
 
+NKikimrKesus::TEvListSemaphoresResult TTestContext::ListSemaphoresFromTablet(bool includeDetails) {
+    const ui64 cookie = RandomNumber<ui64>();
+    const auto edge = Runtime->AllocateEdgeActor();
+    SendFromEdge(edge, new TEvKesus::TEvListSemaphores("", includeDetails), cookie);
+
+    auto result = ExpectEdgeEvent<TEvKesus::TEvListSemaphoresResult>(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<ui64>();
     const auto edge = Runtime->AllocateEdgeActor();

ydb/core/kesus/tablet/ut_helpers.h

@@ -76,6 +76,7 @@ struct TTestContext {
 
     // set/get config requests
     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

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

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"}];
 }

ydb/core/protos/kesus.proto

@@ -139,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;

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);
 }

ydb/public/api/protos/ydb_coordination.proto

@@ -494,3 +494,14 @@ message DescribeNodeResult {
     Ydb.Scheme.Entry self = 1;
     Config config = 2;
 }
+
+/**
+ * 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;
+    Ydb.Operations.OperationParams operation_params = 3;
+}