Merge remote-tracking branch 'fork/feat-implement-TDIGEST.TRIMMED_MEAN-command' into feat-implement-TDIGEST.TRIMMED_MEAN-command

Author: chakkk309

.github/workflows/kvrocks.yaml

@@ -39,7 +39,7 @@ jobs:
     outputs:
       docs_only: ${{ steps.result.outputs.docs_only }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36
         id: changes
         with:
@@ -56,7 +56,7 @@ jobs:
     env:
       FORCE_COLOR: 1
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - name: Check typos
         uses: crate-ci/typos@v1.43.1
         with:
@@ -71,8 +71,8 @@ jobs:
     if: ${{ needs.precondition.outputs.docs_only != 'true' }}
     runs-on: ubuntu-24.04
     steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
+      - uses: actions/checkout@v6
+      - uses: actions/setup-go@v6
         with:
           go-version-file: 'tests/gocase/go.mod'
           cache: false
@@ -96,7 +96,7 @@ jobs:
           git diff -p > clang-format.patch
           cat clang-format.patch
       - name: Upload format patch
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         if: always() && steps.check-format.outcome != 'success'
         with:
           path: clang-format.patch
@@ -291,14 +291,14 @@ jobs:
 
       - name: Cache redis
         id: cache-redis
-        uses: actions/cache@v4
+        uses: actions/cache@v5
         with:
           path: |
             ~/local/bin/redis-cli
           key: ${{ runner.os }}-${{ runner.arch }}-redis-cli
       - name: Cache redis server
         id: cache-redis-server
-        uses: actions/cache@v4
+        uses: actions/cache@v5
         with:
           path: |
             ~/local/bin/redis-server
@@ -312,14 +312,14 @@ jobs:
           pushd redis-6.2.14 && BUILD_TLS=yes make -j$NPROC redis-cli && mv src/redis-cli $HOME/local/bin/ && popd
           pushd redis-6.2.14 && BUILD_TLS=yes make -j$NPROC redis-server && mv src/redis-server $HOME/local/bin/ && popd
 
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
         with:
           fetch-depth: 0
-      - uses: actions/setup-python@v5
+      - uses: actions/setup-python@v6
         if: ${{ !matrix.arm_linux }}
         with:
           python-version: 3.x
-      - uses: actions/setup-go@v5
+      - uses: actions/setup-go@v6
         with:
           go-version-file: 'tests/gocase/go.mod'
           cache: false
@@ -425,7 +425,7 @@ jobs:
             exit 1
           fi
 
-      - uses: actions/upload-artifact@v4
+      - uses: actions/upload-artifact@v7
         if: ${{ failure() && startsWith(matrix.os, 'ubuntu') }}
         with:
           name: kvrocks-coredumps-${{ matrix.name }}
@@ -447,7 +447,7 @@ jobs:
       
       - name: Upload SonarCloud data
         if: ${{ matrix.sonarcloud }}
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         with:
           name: sonarcloud-data
           path: |
@@ -463,7 +463,7 @@ jobs:
       matrix:
         os: [ubuntu-24.04, ubuntu-24.04-arm]
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - name: Get core numbers
         run: echo "NPROC=$(nproc)" >> $GITHUB_ENV
       - uses: docker/build-push-action@v6
@@ -591,15 +591,15 @@ jobs:
 
       - name: Cache redis
         id: cache-redis
-        uses: actions/cache@v4
+        uses: actions/cache@v5
         with:
           path: |
             ~/local/bin/redis-cli
           key: ${{ matrix.image }}-redis-cli
 
       - name: Cache redis server
         id: cache-redis-server
-        uses: actions/cache@v4
+        uses: actions/cache@v5
         with:
           path: |
             ~/local/bin/redis-server
@@ -615,8 +615,8 @@ jobs:
           pushd redis-6.2.14 && USE_JEMALLOC=no make -j$NPROC redis-cli && mv src/redis-cli $HOME/local/bin/ && popd
           pushd redis-6.2.14 && USE_JEMALLOC=no make -j$NPROC redis-server && mv src/redis-server $HOME/local/bin/ && popd
 
-      - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
+      - uses: actions/checkout@v6
+      - uses: actions/setup-go@v6
         if: ${{ !startsWith(matrix.image, 'opensuse') }}
         with:
           go-version-file: 'tests/gocase/go.mod'

.github/workflows/nightly.yaml

@@ -42,7 +42,7 @@ jobs:
           platform=${{ matrix.platform }}
           echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
 
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
         with:
           fetch-depth: 0
 
@@ -60,7 +60,7 @@ jobs:
           password: ${{ secrets.DOCKER_PASSWORD }}
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
+        uses: docker/setup-qemu-action@v4
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
@@ -82,7 +82,7 @@ jobs:
           touch "${{ runner.temp }}/digests/${digest#sha256:}"
 
       - name: Upload digest
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v7
         if: ${{ github.event_name != 'pull_request' }}
         with:
           name: digests-${{ env.PLATFORM_PAIR }}
@@ -98,14 +98,14 @@ jobs:
       - build
     steps:
       - name: Download digests
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v8
         with:
           path: ${{ runner.temp }}/digests
           pattern: digests-*
           merge-multiple: true
 
       - name: Login to Docker Hub
-        uses: docker/login-action@v3
+        uses: docker/login-action@v4
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}

.github/workflows/pr-lint.yaml

@@ -32,7 +32,7 @@ jobs:
     name: Validate PR title
     runs-on: ubuntu-latest
     steps:
-      - uses: amannn/action-semantic-pull-request@v5
+      - uses: amannn/action-semantic-pull-request@v6
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:

.github/workflows/sonar.yaml

@@ -28,15 +28,15 @@ jobs:
     runs-on: ubuntu-22.04
     if: github.event.workflow_run.conclusion == 'success' && github.repository_owner == 'apache'
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
         with:
           repository: ${{ github.event.workflow_run.head_repository.full_name }}
           ref: ${{ github.event.workflow_run.head_sha }}
           fetch-depth: 0
       - name: Install Build Wrapper
-        uses: SonarSource/sonarqube-scan-action/install-build-wrapper@v6.0.0
+        uses: SonarSource/sonarqube-scan-action/install-build-wrapper@v7.0.0
       - name: 'Download code coverage'
-        uses: actions/github-script@v7
+        uses: actions/github-script@v8
         with:
           script: |
             let allArtifacts = await github.rest.actions.listWorkflowRunArtifacts({
@@ -62,7 +62,7 @@ jobs:
           mkdir -p build/CMakeFiles/CMakeTmp
           ls -a sonarcloud-data build
 
-      - uses: actions/setup-python@v5
+      - uses: actions/setup-python@v6
         with:
           python-version: 3.x
 
@@ -73,7 +73,7 @@ jobs:
           echo "The PR number is ${PR_NUMBER:-<none>}"
 
       - name: SonarQube Scan
-        uses: SonarSource/sonarqube-scan-action@v6.0.0
+        uses: SonarSource/sonarqube-scan-action@v7.0.0
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SONAR_TOKEN: ${{ secrets.SONARCLOUD_TOKEN }}

src/commands/cmd_server.cc

@@ -26,6 +26,7 @@
 #include "commander.h"
 #include "commands/scan_base.h"
 #include "common/io_util.h"
+#include "common/logging.h"
 #include "common/rdb_stream.h"
 #include "common/string_util.h"
 #include "common/time_util.h"
@@ -520,6 +521,34 @@ class CommandClient : public Commander {
       return Status::OK();
     }
 
+    if (subcommand_ == "unpause") {
+      if (args.size() != 2) {
+        return {Status::RedisParseErr, errInvalidSyntax};
+      }
+      return Status::OK();
+    }
+
+    if (subcommand_ == "pause") {
+      if (args.size() < 3 || args.size() > 4) {
+        return {Status::RedisParseErr, errInvalidSyntax};
+      }
+      auto parse_result = ParseInt<uint64_t>(args[2], 10);
+      if (!parse_result) {
+        return {Status::RedisParseErr, errValueNotInteger};
+      }
+      pause_timeout_ms_ = *parse_result;
+      pause_mode_ = PauseMode::kAll;
+      if (args.size() == 4) {
+        std::string mode = util::ToLower(args[3]);
+        if (mode == "write") {
+          pause_mode_ = PauseMode::kWrite;
+        } else if (mode != "all") {
+          return {Status::RedisParseErr, errInvalidSyntax};
+        }
+      }
+      return Status::OK();
+    }
+
     if ((subcommand_ == "kill")) {
       if (args.size() == 2) {
         return {Status::RedisParseErr, errInvalidSyntax};
@@ -572,7 +601,9 @@ class CommandClient : public Commander {
       }
       return Status::OK();
     }
-    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY"};
+    return {Status::RedisInvalidCmd,
+            "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY|"
+            "PAUSE|UNPAUSE"};
   }
 
   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
@@ -611,9 +642,28 @@ class CommandClient : public Commander {
         *output = redis::RESP_OK;
       }
       return Status::OK();
+    } else if (subcommand_ == "pause") {
+      if (!conn->IsAdmin()) {
+        return {Status::RedisExecErr, errAdminPermissionRequired};
+      }
+      uint64_t now_ms = util::GetTimeStampMS();
+      srv->PauseConns(now_ms + pause_timeout_ms_, pause_mode_);
+      WARN("CLIENT PAUSE executed, timeout={}ms, mode={}, addr: {}", pause_timeout_ms_,
+           pause_mode_ == PauseMode::kWrite ? "write" : "all", conn->GetAddr());
+      *output = redis::RESP_OK;
+      return Status::OK();
+    } else if (subcommand_ == "unpause") {
+      if (!conn->IsAdmin()) {
+        return {Status::RedisExecErr, errAdminPermissionRequired};
+      }
+      srv->UnpauseConns();
+      *output = redis::RESP_OK;
+      return Status::OK();
     }
 
-    return {Status::RedisInvalidCmd, "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY"};
+    return {Status::RedisInvalidCmd,
+            "Syntax error, try CLIENT LIST|INFO|KILL ip:port|GETNAME|SETNAME|REPLY|"
+            "PAUSE|UNPAUSE"};
   }
 
  private:
@@ -625,6 +675,8 @@ class CommandClient : public Commander {
   int64_t kill_type_ = 0;
   uint64_t id_ = 0;
   bool new_format_ = true;
+  uint64_t pause_timeout_ms_ = 0;
+  PauseMode pause_mode_ = PauseMode::kAll;
 };
 
 class CommandMonitor : public Commander {

src/server/redis_connection.cc

@@ -209,6 +209,7 @@ std::string Connection::GetFlags() const {
   if (IsFlagEnabled(kMonitor)) flags.append("M");
   if (IsFlagEnabled(kAsking)) flags.append("A");
   if (!subscribe_channels_.empty() || !subscribe_patterns_.empty()) flags.append("P");
+  if (is_paused_) flags.append("z");
   if (flags.empty()) flags = "N";
   return flags;
 }
@@ -226,6 +227,21 @@ bool Connection::CanMigrate() const {
          && subscribe_channels_.empty() && subscribe_patterns_.empty();  // not subscribing any channel
 }
 
+void Connection::Pause() {
+  if (is_paused_) return;
+  is_paused_ = true;
+  bufferevent_disable(bev_, EV_READ);
+}
+
+void Connection::Unpause() {
+  if (!is_paused_) return;
+  is_paused_ = false;
+  bufferevent_enable(bev_, EV_READ);
+  // Trigger OnRead so commands buffered in req_ during the pause are processed.
+  // Without this, no new data arrives on the socket and OnRead would never fire.
+  bufferevent_trigger(bev_, EV_READ, BEV_TRIG_IGNORE_WATERMARKS);
+}
+
 void Connection::SubscribeChannel(const std::string &channel) {
   for (const auto &chan : subscribe_channels_) {
     if (channel == chan) return;
@@ -451,6 +467,14 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
     }
 
     auto cmd_flags = attributes->GenerateFlags(cmd_tokens, *config);
+
+    // Push the command back and stop processing; it will be re-executed after unpause.
+    if (srv_->PauseConnIfNeeded(this, cmd_name, cmd_flags)) {
+      multi_error_exit.Disable();  // Don't mark transaction as failed - we're deferring, not erroring
+      to_process_cmds->push_front(std::move(cmd_tokens));
+      return;
+    }
+
     if (GetNamespace().empty()) {
       if (!password.empty()) {
         if (!(cmd_flags & kCmdAuth)) {

src/server/redis_connection.h

@@ -176,6 +176,11 @@ class Connection : public EvbufCallbackBase<Connection> {
   bool IsImporting() const { return importing_; }
   bool CanMigrate() const;
 
+  // CLIENT PAUSE async suspend/resume
+  void Pause();
+  void Unpause();
+  bool IsPaused() const { return is_paused_; }
+
   // Multi exec
   void SetInExec() { in_exec_ = true; }
   bool IsInExec() const { return in_exec_; }
@@ -230,6 +235,8 @@ class Connection : public EvbufCallbackBase<Connection> {
 
   ReplyMode reply_mode_ = ReplyMode::ON;
   std::vector<std::string> queued_replies_;
+
+  bool is_paused_ = false;
 };
 
 }  // namespace redis