Merge branch 'stable-25-1' into fix_update_muted_ya_stable_25_1

Author: naspirato

.github/workflows/nightly_build.yml

@@ -1,27 +1,43 @@
 name: Nightly-Build  # workflow used to upload built binaries to s3
 on:
+  schedule:
+    - cron: "0 3 * * *"  # At 03:00 every day
   workflow_dispatch:
     inputs:
-      runner_label:
-        type: string
-        default: "auto-provisioned"
-        description: "runner label"
-      commit_sha:
-        type: string
-        default: ""
+      use_default_branches:
+        description: 'If true, run on main and all current stable branches. If false, run only on the current branch.'
+        type: boolean
+        required: false
+        default: true
+
 jobs:
+  determine_branches:
+    runs-on: ubuntu-latest
+    outputs:
+      branches: ${{ steps.set-branches.outputs.branches }}
+    steps:
+      - id: set-branches
+        run: |
+          if [[ "${{ github.event_name }}" == "schedule" || "${{ inputs.use_default_branches }}" == "true" ]]; then
+            echo "branches=['main', 'stable-25-1', 'stable-24-4', 'stable-25-1-1', 'stable-25-1-analytics']" >> $GITHUB_OUTPUT
+          else
+            echo "branches=['${{ github.ref_name }}']" >> $GITHUB_OUTPUT
+          fi
+
   build_and_test:
+    needs: determine_branches
     strategy:
       fail-fast: false
       matrix:
         build_preset: ["release", "relwithdebinfo", "release-asan"]
+        branch: ${{ fromJSON(needs.determine_branches.outputs.branches) }}
     runs-on: [ self-hosted, auto-provisioned, "${{ format('build-preset-{0}', matrix.build_preset) }}" ]
-    name: Build and test ${{ matrix.build_preset }}
+    name: Build and test ${{ matrix.build_preset }} on ${{ matrix.branch }}
     steps:
     - name: Checkout
       uses: actions/checkout@v4
       with:
-        ref: ${{ inputs.commit_sha }}
+        ref: ${{ matrix.branch }}
         fetch-depth: 2
     - name: Setup ydb access
       uses: ./.github/actions/setup_ci_ydb_service_account_key_file_credentials
@@ -51,4 +67,4 @@ jobs:
       shell: bash
       run: |
         set -x
-        s3cmd sync --follow-symlinks --acl-public --no-progress --stats --no-check-md5 "ydb/apps/ydbd/ydbd" "s3://ydb-builds/${{ github.ref_name }}/${{ matrix.build_preset }}/ydbd" -d
+        s3cmd sync --follow-symlinks --acl-public --no-progress --stats --no-check-md5 "ydb/apps/ydbd/ydbd" "s3://ydb-builds/${{ matrix.branch }}/${{ matrix.build_preset }}/ydbd" -d

.github/workflows/run_tests.yml

@@ -1,4 +1,4 @@
-name: Run tests
+name: Run-tests
 
 on:
   workflow_call:
@@ -68,10 +68,6 @@ on:
           - release-msan
           - release-tsan
         default: relwithdebinfo
-      branches:
-        description: 'Branches to test (JSON array or single branch name)'
-        required: false
-        default: '["main"]'
 
 jobs:
   prepare:
@@ -81,17 +77,27 @@ jobs:
     steps:
       - name: Set branches
         id: set-branches
+        env:
+          CALLED_BRANCHES: '${{ inputs.branches }}'
         run: |
-          INPUT_BRANCHES='${{ inputs.branches }}'
-          # Check if input is a JSON array
-          if [[ $INPUT_BRANCHES == \[* ]]; then
-            echo "branch_array=$INPUT_BRANCHES" >> $GITHUB_OUTPUT
+          # Проверяем, был ли передан параметр branches из вызывающего workflow
+          if [[ -n "$CALLED_BRANCHES" ]]; then
+            echo "Branches parameter provided from calling workflow: $CALLED_BRANCHES"
+            
+            # Проверяем, является ли вход уже JSON-массивом
+            if [[ $CALLED_BRANCHES == \[* ]]; then
+              echo "branch_array=$CALLED_BRANCHES" >> $GITHUB_OUTPUT
+            else
+              # Если это одна ветка, создаем JSON-массив с одним элементом
+              echo "branch_array=[\"$CALLED_BRANCHES\"]" >> $GITHUB_OUTPUT
+            fi
           else
-            # If it's a single branch, create a JSON array with one element
-            echo "branch_array=[\"$INPUT_BRANCHES\"]" >> $GITHUB_OUTPUT
+            # Если ветки не переданы, значит это прямой запуск workflow_dispatch
+            echo "No branches specified, using current branch: ${{ github.ref_name }}"
+            echo "branch_array=[\"${{ github.ref_name }}\"]" >> $GITHUB_OUTPUT
           fi
           
-          echo "Using branches: $(cat $GITHUB_OUTPUT | grep branch_array | cut -d= -f2)"
+          echo "Final branches to use: $(cat $GITHUB_OUTPUT | grep branch_array | cut -d= -f2)"
 
   run_tests:
     needs: prepare

.gitignore

@@ -88,3 +88,7 @@ bin/config.json
 
 # handy for local junk, which is not intended to appear in the repo
 junk/
+
+# YDB CLI configuration file location is `~/ydb/config/config.yaml`
+config/config.yaml
+path.bash.inc

ydb/core/blobstorage/nodewarden/node_warden_impl.cpp

@@ -1528,7 +1528,7 @@ bool NKikimr::NStorage::DeriveStorageConfig(const NKikimrConfig::TAppConfig& app
 
             // find state storage setup for that domain
             for (const auto& ss : domains.GetStateStorage()) {
-                if (domain.SSIdSize() == 1 && ss.GetSSId() == domain.GetSSId(0)) {
+                if (domain.SSIdSize() == 0 || (domain.SSIdSize() == 1 && ss.GetSSId() == domain.GetSSId(0))) {
                     const bool hadStateStorageConfig = config->HasStateStorageConfig();
                     const bool hadStateStorageBoardConfig = config->HasStateStorageBoardConfig();
                     const bool hadSchemeBoardConfig = config->HasSchemeBoardConfig();

ydb/core/driver_lib/run/run.cpp

@@ -946,6 +946,40 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
         opts.SetMaxMessageSize(grpcConfig.HasMaxMessageSize() ? grpcConfig.GetMaxMessageSize() : NYdbGrpc::DEFAULT_GRPC_MESSAGE_SIZE_LIMIT);
         opts.SetMaxGlobalRequestInFlight(grpcConfig.GetMaxInFlight());
         opts.SetLogger(NYdbGrpc::CreateActorSystemLogger(*ActorSystem.Get(), NKikimrServices::GRPC_SERVER));
+        switch(grpcConfig.GetDefaultCompressionAlgorithm()) {
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_NONE: {
+                opts.SetDefaultCompressionAlgorithm(GRPC_COMPRESS_NONE);
+                break;
+            }
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_DEFLATE: {
+                opts.SetDefaultCompressionAlgorithm(GRPC_COMPRESS_DEFLATE);
+                break;
+            }
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_GZIP: {
+                opts.SetDefaultCompressionAlgorithm(GRPC_COMPRESS_GZIP);
+                break;
+            }
+        }
+
+        switch(grpcConfig.GetDefaultCompressionLevel()) {
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_LEVEL_NONE: {
+                opts.SetDefaultCompressionLevel(GRPC_COMPRESS_LEVEL_NONE);
+                break;
+            }
+
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_LEVEL_LOW: {
+                opts.SetDefaultCompressionLevel(GRPC_COMPRESS_LEVEL_LOW);
+                break;
+            }
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_LEVEL_MED: {
+                opts.SetDefaultCompressionLevel(GRPC_COMPRESS_LEVEL_MED);
+                break;
+            }
+            case NKikimrConfig::TGRpcConfig::YDB_GRPC_COMPRESS_LEVEL_HIGH: {
+                opts.SetDefaultCompressionLevel(GRPC_COMPRESS_LEVEL_HIGH);
+                break;
+            }
+        }
 
         if (appConfig.HasDomainsConfig() &&
             appConfig.GetDomainsConfig().HasSecurityConfig() &&

ydb/core/grpc_services/rpc_read_rows.cpp

@@ -732,6 +732,14 @@ class TReadRowsRPC : public TActorBootstrapped<TReadRowsRPC> {
         SendResult(status, errorMsg, issues);
     }
 
+    void Handle(TEvents::TEvUndelivered::TPtr&) {
+        return ReplyWithError(Ydb::StatusIds::INTERNAL_ERROR, "Internal error: pipe cache is not available, the cluster might not be configured properly");
+    }
+
+    void Handle(TEvPipeCache::TEvDeliveryProblem::TPtr &ev) {
+        return ReplyWithError(Ydb::StatusIds::UNAVAILABLE, TStringBuilder() << "Failed to connect to shard " << ev->Get()->TabletId);
+    }
+
     void PassAway() override {
         Send(PipeCache, new TEvPipeCache::TEvUnlink(0));
         if (TimeoutTimerActorId) {
@@ -748,6 +756,9 @@ class TReadRowsRPC : public TActorBootstrapped<TReadRowsRPC> {
             hFunc(TEvTxProxySchemeCache::TEvResolveKeySetResult, Handle);
             hFunc(TEvDataShard::TEvReadResult, Handle);
 
+            hFunc(TEvents::TEvUndelivered, Handle);
+            hFunc(TEvPipeCache::TEvDeliveryProblem, Handle);
+
             hFunc(TEvents::TEvWakeup, HandleTimeout);
         }
     }

ydb/core/persqueue/pq_impl.cpp

@@ -2904,9 +2904,41 @@ void TPersQueue::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev, const TActo
         return;
     }
 
+    if (ev->Get()->Dead) {
+        AckReadSetsToTablet(ev->Get()->TabletId, ctx);
+        return;
+    }
+
     RestartPipe(ev->Get()->TabletId, ctx);
 }
 
+void TPersQueue::AckReadSetsToTablet(ui64 tabletId, const TActorContext& ctx)
+{
+    THashSet<TDistributedTransaction*> txs;
+
+    for (ui64 txId : GetBindedTxs(tabletId)) {
+        auto* tx = GetTransaction(ctx, txId);
+        if (!tx) {
+            continue;
+        }
+
+        tx->OnReadSetAck(tabletId);
+        tx->UnbindMsgsFromPipe(tabletId);
+
+        txs.insert(tx);
+    }
+
+    if (txs.empty()) {
+        return;
+    }
+
+    for (auto* tx : txs) {
+        TryExecuteTxs(ctx, *tx);
+    }
+
+    TryWriteTxs(ctx);
+}
+
 void TPersQueue::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActorContext& ctx)
 {
     PQ_LOG_D("Handle TEvTabletPipe::TEvClientDestroyed");
@@ -2919,7 +2951,7 @@ void TPersQueue::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActo
 
 void TPersQueue::RestartPipe(ui64 tabletId, const TActorContext& ctx)
 {
-    for (auto& txId: GetBindedTxs(tabletId)) {
+    for (ui64 txId : GetBindedTxs(tabletId)) {
         auto* tx = GetTransaction(ctx, txId);
         if (!tx) {
             continue;

ydb/core/persqueue/pq_impl.h

@@ -566,6 +566,8 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
 
     void AddPendingEvent(IEventHandle* ev);
     void ProcessPendingEvents();
+
+    void AckReadSetsToTablet(ui64 tabletId, const TActorContext& ctx);
 };
 
 

ydb/core/persqueue/transaction.cpp

@@ -312,8 +312,13 @@ void TDistributedTransaction::OnReadSetAck(const NKikimrTx::TEvReadSetAck& event
     Y_ABORT_UNLESS(event.HasStep() && (Step == event.GetStep()));
     Y_ABORT_UNLESS(event.HasTxId() && (TxId == event.GetTxId()));
 
-    if (PredicateRecipients.contains(event.GetTabletConsumer())) {
-        PredicateRecipients[event.GetTabletConsumer()] = true;
+    OnReadSetAck(event.GetTabletConsumer());
+}
+
+void TDistributedTransaction::OnReadSetAck(ui64 tabletId)
+{
+    if (PredicateRecipients.contains(tabletId)) {
+        PredicateRecipients[tabletId] = true;
         ++PredicateAcksCount;
 
         PQ_LOG_D("Predicate acks " << PredicateAcksCount << "/" << PredicateRecipients.size());

ydb/core/persqueue/transaction.h

@@ -34,6 +34,7 @@ struct TDistributedTransaction {
                    const TActorId& sender,
                    std::unique_ptr<TEvTxProcessing::TEvReadSetAck> ack);
     void OnReadSetAck(const NKikimrTx::TEvReadSetAck& event);
+    void OnReadSetAck(ui64 tabletId);
     void OnTxCommitDone(const TEvPQ::TEvTxCommitDone& event);
 
     using EDecision = NKikimrTx::TReadSetData::EDecision;