feat: add Redis Vector Set support

.cargo/audit.toml

@@ -0,0 +1,31 @@
+# Copyright (c) 2024-present, arana-db Community.  All rights reserved.
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# cargo audit configuration. This complements the [advisories] ignore
+# governance kept in deny.toml. Every ignored advisory must carry a reason
+# and a remove_when condition (documented in the comment next to it) and must
+# be re-audited once that condition is met.
+
+[advisories]
+ignore = [
+  # RUSTSEC-2026-0235: rkyv 0.7.46 is a transitive dependency of openraft
+  # 0.9.25 (Raft wire serialization); no safe in-scope upgrade exists below a
+  # breaking openraft bump. The advisory requires a maliciously crafted
+  # archive with Rc/Arc and is not reachable from the cluster's own binlog
+  # writes. Remove this exception once openraft pulls rkyv >= 0.8.17.
+  "RUSTSEC-2026-0235",
+]

.github/workflows/ci.yml

@@ -310,17 +310,23 @@ jobs:
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
           toolchain: ${{ env.SANITIZER_TOOLCHAIN }}
-          components: rust-src
+          components: |
+            rust-src
+            llvm-tools-preview
 
       - name: Install protoc
         run: |
           sudo apt-get update
-          sudo apt-get install -y protobuf-compiler
+          sudo apt-get install -y protobuf-compiler llvm
 
       - name: Run tests with ${{ matrix.sanitizer }} sanitizer
         env:
           RUSTFLAGS: "-Z sanitizer=${{ matrix.sanitizer }}"
           RUSTDOCFLAGS: "-Z sanitizer=${{ matrix.sanitizer }}"
+          # Keep source locations in sanitizer backtraces. The repository-wide
+          # default is optimized for normal CI logs, but an unsymbolized LSan
+          # report cannot identify the leaking test/helper.
+          CARGO_PROFILE_DEV_DEBUG: 2
           # AddressSanitizer enables leak detection by default. Leak checks run in
           # the dedicated leak sanitizer job, where third-party suppressions are
           # applied explicitly.
@@ -329,7 +335,34 @@ jobs:
           TSAN_OPTIONS: "suppressions=${{ github.workspace }}/.github/tsan.supp"
         run: |
           set -o pipefail
-          if [[ "${{ matrix.sanitizer }}" == "leak" || "${{ matrix.sanitizer }}" == "thread" ]]; then
+          sanitizer_sysroot="$(rustc +${SANITIZER_TOOLCHAIN} --print sysroot)"
+          rust_llvm_symbolizer="$(find "$sanitizer_sysroot" -type f -name llvm-symbolizer -perm -111 -print -quit 2>/dev/null || true)"
+          if [[ -n "$rust_llvm_symbolizer" ]]; then
+            export ASAN_SYMBOLIZER_PATH="$rust_llvm_symbolizer"
+          elif command -v llvm-symbolizer >/dev/null 2>&1; then
+            export ASAN_SYMBOLIZER_PATH="$(command -v llvm-symbolizer)"
+          fi
+          if [[ -n "${ASAN_SYMBOLIZER_PATH:-}" ]]; then
+            export LSAN_OPTIONS="${LSAN_OPTIONS}:external_symbolizer_path=${ASAN_SYMBOLIZER_PATH}:symbolize=1:malloc_context_size=50"
+            echo "Using sanitizer symbolizer: $ASAN_SYMBOLIZER_PATH"
+            "$ASAN_SYMBOLIZER_PATH" --version || true
+          fi
+          if [[ "${{ matrix.sanitizer }}" == "leak" ]]; then
+            # Run the complete leak-sanitizer scope after the symbolized
+            # diagnostic experiments; this is the final validation for the PR.
+            cargo +${SANITIZER_TOOLCHAIN} test -Z build-std --target x86_64-unknown-linux-gnu \
+              --workspace --exclude storage \
+              -- --skip cursor_snapshot_roundtrip \
+              --skip install_snapshot_with_existing_data \
+              --skip install_snapshot_rearms_append_log_hook \
+              --skip test_snapshot_with_logindex_state \
+              --skip test_on_binlog_write_updates_collector \
+              --skip test_collector_state_export_restore \
+              --skip storage_command_e2e_ \
+              2>&1 | tee sanitizer.log
+            cargo +${SANITIZER_TOOLCHAIN} test -Z build-std --target x86_64-unknown-linux-gnu \
+              -p storage --lib 2>&1 | tee -a sanitizer.log
+          elif [[ "${{ matrix.sanitizer }}" == "thread" ]]; then
             # RocksDB/libstdc++ trigger sanitizer reports and timing instability
             # in storage integration test binaries. The net crate's end-to-end
             # storage command tests also spin up a real RocksDB-backed stack, so

docs/cluster.md

@@ -60,6 +60,8 @@ Raft 日志始终持久化到 RocksDB。节点重启时会从 `raft-data-dir` 
 
 升级说明：曾启用 `raft-use-memory-log-store` 的节点不能保留原 node ID 和主数据原地升级。内存日志模式没有持久化 vote、membership、committed position 和日志尾部，无法从主数据安全恢复完整的 Raft 状态。只要 `raft-data-dir` 中仍存在旧 `raft_logs` 目录，节点就会拒绝启动，即使同时存在 `raft_logs_rocksdb` 状态，因为无法证明两者属于同一段完整 Raft 历史。请为该节点使用新的 node ID 和干净的 `data-dir`/`raft-data-dir`，再从健康 leader 重新加入集群。这个保守门禁也会阻止仅留下空旧目录的节点；确认不需要旧数据后，应按同样流程清理并重新入群，而不是删除单个标记目录后复用原 node ID。
 
+快照格式 v1 是未发布的开发期格式，当前 v2 节点会明确拒绝它，不支持 v1→v2 滚动兼容。仍保留 v1 快照或由 v1 状态压缩过日志的节点必须停机后使用新的 node ID 和干净的 `data-dir`/`raft-data-dir`，再从健康的 v2 leader 重建并重新入群。
+
 ### 2. 启动节点
 
 ```bash

src/cmd/src/admin.rs

@@ -26,6 +26,29 @@ use storage::storage::Storage;
 
 use crate::{AclCategory, Cmd, CmdFlags, CmdMeta, impl_cmd_clone_box, impl_cmd_meta};
 
+/// INFO VECTOR section: the Phase 1 index kind plus the FLAT query counters.
+/// vector_sets / vector_elements are omitted on purpose: counting them needs
+/// a full keyspace scan and INFO must stay O(1).
+fn vector_section(storage: &Storage) -> String {
+    let metrics = storage.vector_metrics();
+    format!(
+        "# Vector\r\n\
+         index-kind:flat\r\n\
+         vector_flat_queries_total:{}\r\n\
+         vector_flat_query_timeouts_total:{}\r\n\
+         vector_flat_query_errors_total:{}\r\n\
+         vector_search_capacity_rejected_total:{}\r\n\
+         vector_flat_query_duration_micros_total:{}\r\n\
+         vector_flat_query_duration_count:{}\r\n",
+        metrics.flat_queries_total,
+        metrics.flat_query_timeouts_total,
+        metrics.flat_query_errors_total,
+        metrics.capacity_rejected_total,
+        metrics.flat_query_duration_micros_total,
+        metrics.flat_query_duration_count,
+    )
+}
+
 /// INFO command - Show server information including cluster status
 #[derive(Clone, Default)]
 pub struct InfoCmd {
@@ -54,7 +77,7 @@ impl Cmd for InfoCmd {
         true
     }
 
-    fn do_cmd(&self, client: &Client, _storage: Arc<Storage>) {
+    fn do_cmd(&self, client: &Client, storage: Arc<Storage>) {
         let section = if client.argv().len() > 1 {
             String::from_utf8_lossy(&client.argv()[1]).to_lowercase()
         } else {
@@ -69,6 +92,9 @@ impl Cmd for InfoCmd {
                 info.push_str("cluster_enabled:0\r\n");
                 info.push_str("cluster_state:disabled\r\n");
             }
+            "vector" => {
+                info.push_str(&vector_section(&storage));
+            }
             "server" | "default" => {
                 info.push_str("# Server\r\n");
                 info.push_str("redis_version:7.0.0\r\n");
@@ -95,6 +121,8 @@ impl Cmd for InfoCmd {
                     info.push_str("\r\n# Cluster\r\n");
                     info.push_str("cluster_enabled:0\r\n");
                     info.push_str("cluster_state:disabled\r\n");
+                    info.push_str("\r\n");
+                    info.push_str(&vector_section(&storage));
                 }
             }
             _ => {

src/cmd/src/auth.rs

@@ -27,6 +27,14 @@ use crate::{impl_cmd_clone_box, impl_cmd_meta};
 
 pub type RequirepassProvider = Arc<dyn Fn() -> Option<String> + Send + Sync>;
 
+fn no_requirepass() -> Option<String> {
+    None
+}
+
+pub(crate) fn no_requirepass_provider() -> RequirepassProvider {
+    Arc::new(no_requirepass as fn() -> Option<String>)
+}
+
 #[derive(Clone)]
 pub struct AuthCmd {
     meta: CmdMeta,
@@ -43,7 +51,7 @@ impl Default for AuthCmd {
                 acl_category: AclCategory::CONNECTION,
                 ..Default::default()
             },
-            requirepass_provider: Arc::new(|| None),
+            requirepass_provider: no_requirepass_provider(),
         }
     }
 }
@@ -118,7 +126,9 @@ mod tests {
     use client::StreamTrait;
     use storage::storage::Storage;
 
-    struct TestStream;
+    struct TestStream {
+        _marker: u8,
+    }
 
     #[async_trait::async_trait]
     impl StreamTrait for TestStream {
@@ -133,7 +143,7 @@ mod tests {
 
     fn make_client() -> Arc<Client> {
         // `Client::new` is fail-closed: a freshly built client is unauthenticated.
-        Arc::new(Client::new(Box::new(TestStream)))
+        Arc::new(Client::new(Box::new(TestStream { _marker: 0 })))
     }
 
     fn make_storage() -> Arc<Storage> {
@@ -223,4 +233,11 @@ mod tests {
         let cmd = AuthCmd::default();
         assert!(cmd.has_flag(CmdFlags::NO_AUTH));
     }
+
+    #[test]
+    fn default_requirepass_provider_has_nonzero_trait_object_data_size() {
+        let cmd = AuthCmd::default();
+
+        assert_ne!(std::mem::size_of_val(cmd.requirepass_provider.as_ref()), 0);
+    }
 }

src/cmd/src/hello.rs

@@ -23,6 +23,7 @@ use resp::{CommandType, HelloAuthResult, RespCommand, RespData, RespError};
 use storage::storage::Storage;
 use subtle::ConstantTimeEq;
 
+use crate::auth::no_requirepass_provider;
 use crate::{
     AclCategory, Cmd, CmdFlags, CmdMeta, RequirepassProvider, impl_cmd_clone_box, impl_cmd_meta,
 };
@@ -43,7 +44,7 @@ impl Default for HelloCmd {
                 acl_category: AclCategory::CONNECTION | AclCategory::FAST,
                 ..Default::default()
             },
-            requirepass_provider: Arc::new(|| None),
+            requirepass_provider: no_requirepass_provider(),
         }
     }
 }
@@ -138,7 +139,9 @@ mod tests {
     use super::*;
     use client::StreamTrait;
 
-    struct TestStream;
+    struct TestStream {
+        _marker: u8,
+    }
 
     #[async_trait::async_trait]
     impl StreamTrait for TestStream {
@@ -152,7 +155,7 @@ mod tests {
     }
 
     fn make_client() -> Arc<Client> {
-        Arc::new(Client::new(Box::new(TestStream)))
+        Arc::new(Client::new(Box::new(TestStream { _marker: 0 })))
     }
 
     fn make_storage() -> Arc<Storage> {
@@ -226,4 +229,11 @@ mod tests {
 
         assert_eq!(client.name().as_slice(), b"my-client");
     }
+
+    #[test]
+    fn default_requirepass_provider_has_nonzero_trait_object_data_size() {
+        let cmd = HelloCmd::default();
+
+        assert_ne!(std::mem::size_of_val(cmd.requirepass_provider.as_ref()), 0);
+    }
 }

src/cmd/src/hscan.rs

@@ -112,7 +112,9 @@ mod tests {
 
     use super::*;
 
-    struct TestStream;
+    struct TestStream {
+        _marker: u8,
+    }
 
     #[async_trait::async_trait]
     impl StreamTrait for TestStream {
@@ -175,7 +177,7 @@ mod tests {
             .unwrap();
         storage.hset(b"binary_hash", b"\xff", b"\xfe").unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         client.set_argv(&[
             b"hscan".to_vec(),
             b"binary_hash".to_vec(),
@@ -211,7 +213,7 @@ mod tests {
             .unwrap();
         storage.hset(b"empty_field_hash", b"", b"value").unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         client.set_argv(&[
             b"hscan".to_vec(),
             b"empty_field_hash".to_vec(),

src/cmd/src/keys.rs

@@ -89,7 +89,9 @@ mod tests {
     use super::KeysCmd;
     use crate::Cmd;
 
-    struct TestStream;
+    struct TestStream {
+        _marker: u8,
+    }
 
     #[async_trait::async_trait]
     impl StreamTrait for TestStream {
@@ -142,7 +144,7 @@ mod tests {
         storage.set(&invalid_key_a, b"invalid-a").unwrap();
         storage.set(&invalid_key_b, b"invalid-b").unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         let command = KeysCmd::new();
 
         client.set_argv(&[b"keys".to_vec(), b"?".to_vec()]);
@@ -173,7 +175,7 @@ mod tests {
         storage.set(b"", b"empty-key").unwrap();
         storage.set(b"literal", b"literal-key").unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         let command = KeysCmd::new();
 
         assert_eq!(

src/cmd/src/lib.rs

@@ -93,6 +93,7 @@ pub mod sunionstore;
 pub mod table;
 pub mod ttl;
 pub mod type_cmd;
+pub mod vector;
 pub mod zadd;
 pub mod zcard;
 pub mod zcount;
@@ -192,6 +193,11 @@ pub trait Cmd: Send + Sync {
 
     fn do_initial(&self, client: &Client) -> bool;
 
+    /// Reject commands that must not reach cluster routing or read barriers.
+    fn check_pre_route(&self, _client: &Client) -> bool {
+        true
+    }
+
     fn do_cmd(&self, client: &Client, storage: Arc<Storage>);
 
     fn clone_box(&self) -> Box<dyn Cmd>;

src/cmd/src/sscan.rs

@@ -110,7 +110,9 @@ mod tests {
 
     use super::*;
 
-    struct TestStream;
+    struct TestStream {
+        _marker: u8,
+    }
 
     #[async_trait::async_trait]
     impl StreamTrait for TestStream {
@@ -173,7 +175,7 @@ mod tests {
             .unwrap();
         storage.sadd(b"binary_set", &[b"\xff"]).unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         client.set_argv(&[
             b"sscan".to_vec(),
             b"binary_set".to_vec(),
@@ -206,7 +208,7 @@ mod tests {
             .unwrap();
         storage.sadd(b"empty_member_set", &[b""]).unwrap();
         let storage = Arc::new(storage);
-        let client = Client::new(Box::new(TestStream));
+        let client = Client::new(Box::new(TestStream { _marker: 0 }));
         client.set_argv(&[
             b"sscan".to_vec(),
             b"empty_member_set".to_vec(),