Add sync API for transactional client

src/lib.rs

@@ -159,6 +159,12 @@ pub use crate::transaction::Client as TransactionClient;
 #[doc(inline)]
 pub use crate::transaction::Snapshot;
 #[doc(inline)]
+pub use crate::transaction::SyncSnapshot;
+#[doc(inline)]
+pub use crate::transaction::SyncTransaction;
+#[doc(inline)]
+pub use crate::transaction::SyncTransactionClient;
+#[doc(inline)]
 pub use crate::transaction::Transaction;
 #[doc(inline)]
 pub use crate::transaction::TransactionOptions;

src/transaction/mod.rs

@@ -12,6 +12,9 @@ pub use client::Client;
 pub(crate) use lock::resolve_locks;
 pub(crate) use lock::HasLocks;
 pub use snapshot::Snapshot;
+pub use sync_client::SyncTransactionClient;
+pub use sync_snapshot::SyncSnapshot;
+pub use sync_transaction::SyncTransaction;
 pub use transaction::CheckLevel;
 #[doc(hidden)]
 pub use transaction::HeartbeatOption;
@@ -28,5 +31,8 @@ pub use lock::LockResolver;
 pub use lock::ResolveLocksContext;
 pub use lock::ResolveLocksOptions;
 mod snapshot;
+mod sync_client;
+mod sync_snapshot;
+mod sync_transaction;
 #[allow(clippy::module_inception)]
 mod transaction;

src/transaction/sync_client.rs

@@ -0,0 +1,253 @@
+use crate::{
+    request::plan::CleanupLocksResult,
+    transaction::{
+        client::Client, sync_snapshot::SyncSnapshot, sync_transaction::SyncTransaction,
+        ResolveLocksOptions,
+    },
+    BoundRange, Config, Result, Timestamp, TransactionOptions,
+};
+use std::sync::Arc;
+
+/// Synchronous TiKV transactional client.
+///
+/// This is a synchronous wrapper around the async [`TransactionClient`](crate::TransactionClient).
+/// All methods block the current thread until completion.
+///
+/// For async operations, use [`TransactionClient`](crate::TransactionClient) instead.
+pub struct SyncTransactionClient {
+    client: Client,
+    runtime: Arc<tokio::runtime::Runtime>,
+}
+
+impl SyncTransactionClient {
+    /// Create a synchronous transactional [`SyncTransactionClient`] and connect to the TiKV cluster.
+    ///
+    /// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
+    /// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
+    /// (include all endpoints, if possible), this helps avoid having a single point of failure.
+    ///
+    /// This is a synchronous version of [`TransactionClient::new`](crate::TransactionClient::new).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::SyncTransactionClient;
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// ```
+    pub fn new<S: Into<String>>(pd_endpoints: Vec<S>) -> Result<Self> {
+        Self::new_with_config(pd_endpoints, Config::default())
+    }
+
+    /// Create a synchronous transactional [`SyncTransactionClient`] with a custom configuration.
+    ///
+    /// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
+    /// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
+    /// (include all endpoints, if possible), this helps avoid having a single point of failure.
+    ///
+    /// This is a synchronous version of [`TransactionClient::new_with_config`](crate::TransactionClient::new_with_config).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::{Config, SyncTransactionClient};
+    /// # use std::time::Duration;
+    /// let client = SyncTransactionClient::new_with_config(
+    ///     vec!["192.168.0.100"],
+    ///     Config::default().with_timeout(Duration::from_secs(60)),
+    /// )
+    /// .unwrap();
+    /// ```
+    pub fn new_with_config<S: Into<String>>(pd_endpoints: Vec<S>, config: Config) -> Result<Self> {
+        let runtime =
+            Arc::new(tokio::runtime::Runtime::new()?);
+        let client = runtime.block_on(Client::new_with_config(pd_endpoints, config))?;
+        Ok(Self { client, runtime })
+    }
+
+    /// Creates a new optimistic [`SyncTransaction`].
+    ///
+    /// Use the transaction to issue requests like [`get`](SyncTransaction::get) or
+    /// [`put`](SyncTransaction::put).
+    ///
+    /// Write operations do not lock data in TiKV, thus the commit request may fail due to a write
+    /// conflict.
+    ///
+    /// This is a synchronous version of [`TransactionClient::begin_optimistic`](crate::TransactionClient::begin_optimistic).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::SyncTransactionClient;
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// let mut transaction = client.begin_optimistic().unwrap();
+    /// // ... Issue some commands.
+    /// transaction.commit().unwrap();
+    /// ```
+    pub fn begin_optimistic(&self) -> Result<SyncTransaction> {
+        let inner = self.runtime.block_on(self.client.begin_optimistic())?;
+        Ok(SyncTransaction::new(inner, Arc::clone(&self.runtime)))
+    }
+
+    /// Creates a new pessimistic [`SyncTransaction`].
+    ///
+    /// Write operations will lock the data until committed, thus commit requests should not suffer
+    /// from write conflicts.
+    ///
+    /// This is a synchronous version of [`TransactionClient::begin_pessimistic`](crate::TransactionClient::begin_pessimistic).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::SyncTransactionClient;
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// let mut transaction = client.begin_pessimistic().unwrap();
+    /// // ... Issue some commands.
+    /// transaction.commit().unwrap();
+    /// ```
+    pub fn begin_pessimistic(&self) -> Result<SyncTransaction> {
+        let inner = self.runtime.block_on(self.client.begin_pessimistic())?;
+        Ok(SyncTransaction::new(inner, Arc::clone(&self.runtime)))
+    }
+
+    /// Create a new customized [`SyncTransaction`].
+    ///
+    /// This is a synchronous version of [`TransactionClient::begin_with_options`](crate::TransactionClient::begin_with_options).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::{SyncTransactionClient, TransactionOptions};
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// let mut transaction = client
+    ///     .begin_with_options(TransactionOptions::default().use_async_commit())
+    ///     .unwrap();
+    /// // ... Issue some commands.
+    /// transaction.commit().unwrap();
+    /// ```
+    pub fn begin_with_options(&self, options: TransactionOptions) -> Result<SyncTransaction> {
+        let inner = self
+            .runtime
+            .block_on(self.client.begin_with_options(options))?;
+        Ok(SyncTransaction::new(inner, Arc::clone(&self.runtime)))
+    }
+
+    /// Create a new read-only [`SyncSnapshot`] at the given [`Timestamp`].
+    ///
+    /// A snapshot is a read-only transaction that reads data as if the snapshot was taken at the
+    /// specified timestamp. It can read operations that happened before the timestamp, but ignores
+    /// operations after the timestamp.
+    ///
+    /// Use snapshots when you need:
+    /// - Consistent reads across multiple operations without starting a full transaction
+    /// - Point-in-time reads at a specific timestamp
+    /// - Read-only access without the overhead of transaction tracking
+    ///
+    /// Unlike transactions, snapshots cannot perform write operations (put, delete, etc.).
+    ///
+    /// This is a synchronous version of [`TransactionClient::snapshot`](crate::TransactionClient::snapshot).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::{SyncTransactionClient, TransactionOptions};
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// let timestamp = client.current_timestamp().unwrap();
+    /// let mut snapshot = client.snapshot(timestamp, TransactionOptions::default());
+    /// 
+    /// // Read data as it existed at the snapshot timestamp
+    /// let value = snapshot.get("key".to_owned()).unwrap();
+    /// ```
+    pub fn snapshot(&self, timestamp: Timestamp, options: TransactionOptions) -> SyncSnapshot {
+        let inner = self.client.snapshot(timestamp, options);
+        SyncSnapshot::new(inner, Arc::clone(&self.runtime))
+    }
+
+    /// Retrieve the current [`Timestamp`].
+    ///
+    /// This is a synchronous version of [`TransactionClient::current_timestamp`](crate::TransactionClient::current_timestamp).
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use tikv_client::SyncTransactionClient;
+    /// let client = SyncTransactionClient::new(vec!["192.168.0.100"]).unwrap();
+    /// let timestamp = client.current_timestamp().unwrap();
+    /// ```
+    pub fn current_timestamp(&self) -> Result<Timestamp> {
+        self.runtime.block_on(self.client.current_timestamp())
+    }
+
+    /// Request garbage collection (GC) of the TiKV cluster.
+    ///
+    /// GC deletes MVCC records whose timestamp is lower than the given `safepoint`. We must guarantee
+    /// that all transactions started before this timestamp had committed. We can keep an active
+    /// transaction list in application to decide which is the minimal start timestamp of them.
+    ///
+    /// For each key, the last mutation record (unless it's a deletion) before `safepoint` is retained.
+    ///
+    /// GC is performed by:
+    /// 1. resolving all locks with timestamp <= `safepoint`
+    /// 2. updating PD's known safepoint
+    ///
+    /// This is a simplified version of [GC in TiDB](https://docs.pingcap.com/tidb/stable/garbage-collection-overview).
+    /// We skip the second step "delete ranges" which is an optimization for TiDB.
+    ///
+    /// This is a synchronous version of [`TransactionClient::gc`](crate::TransactionClient::gc).
+    pub fn gc(&self, safepoint: Timestamp) -> Result<bool> {
+        self.runtime.block_on(self.client.gc(safepoint))
+    }
+
+    /// Clean up all locks in the specified range.
+    ///
+    /// This is a synchronous version of [`TransactionClient::cleanup_locks`](crate::TransactionClient::cleanup_locks).
+    pub fn cleanup_locks(
+        &self,
+        range: impl Into<BoundRange>,
+        safepoint: &Timestamp,
+        options: ResolveLocksOptions,
+    ) -> Result<CleanupLocksResult> {
+        self.runtime
+            .block_on(self.client.cleanup_locks(range, safepoint, options))
+    }
+
+    /// Cleans up all keys in a range and quickly reclaim disk space.
+    ///
+    /// The range can span over multiple regions.
+    ///
+    /// Note that the request will directly delete data from RocksDB, and all MVCC will be erased.
+    ///
+    /// This interface is intended for special scenarios that resemble operations like "drop table" or "drop database" in TiDB.
+    ///
+    /// This is a synchronous version of [`TransactionClient::unsafe_destroy_range`](crate::TransactionClient::unsafe_destroy_range).
+    pub fn unsafe_destroy_range(&self, range: impl Into<BoundRange>) -> Result<()> {
+        self.runtime
+            .block_on(self.client.unsafe_destroy_range(range))
+    }
+
+    /// Scan all locks in the specified range.
+    ///
+    /// This is only available for integration tests.
+    ///
+    /// Note: `batch_size` must be >= expected number of locks.
+    ///
+    /// This is a synchronous version of [`TransactionClient::scan_locks`](crate::TransactionClient::scan_locks).
+    #[cfg(feature = "integration-tests")]
+    pub fn scan_locks(
+        &self,
+        safepoint: &Timestamp,
+        range: impl Into<BoundRange>,
+        batch_size: u32,
+    ) -> Result<Vec<crate::proto::kvrpcpb::LockInfo>> {
+        self.runtime
+            .block_on(self.client.scan_locks(safepoint, range, batch_size))
+    }
+}
+
+impl Clone for SyncTransactionClient {
+    fn clone(&self) -> Self {
+        Self {
+            client: self.client.clone(),
+            runtime: Arc::clone(&self.runtime),
+        }
+    }
+}

src/transaction/sync_snapshot.rs

@@ -0,0 +1,72 @@
+use crate::{BoundRange, Key, KvPair, Result, Snapshot, Value};
+use std::sync::Arc;
+
+/// A synchronous read-only snapshot.
+///
+/// This is a wrapper around the async [`Snapshot`] that provides blocking methods.
+/// All operations block the current thread until completed.
+pub struct SyncSnapshot {
+    inner: Snapshot,
+    runtime: Arc<tokio::runtime::Runtime>,
+}
+
+impl SyncSnapshot {
+    pub(crate) fn new(inner: Snapshot, runtime: Arc<tokio::runtime::Runtime>) -> Self {
+        Self { inner, runtime }
+    }
+
+    /// Get the value associated with the given key.
+    pub fn get(&mut self, key: impl Into<Key>) -> Result<Option<Value>> {
+        self.runtime.block_on(self.inner.get(key))
+    }
+
+    /// Check whether the key exists.
+    pub fn key_exists(&mut self, key: impl Into<Key>) -> Result<bool> {
+        self.runtime.block_on(self.inner.key_exists(key))
+    }
+
+    /// Get the values associated with the given keys.
+    pub fn batch_get(
+        &mut self,
+        keys: impl IntoIterator<Item = impl Into<Key>>,
+    ) -> Result<impl Iterator<Item = KvPair>> {
+        self.runtime.block_on(self.inner.batch_get(keys))
+    }
+
+    /// Scan a range, return at most `limit` key-value pairs that lying in the range.
+    pub fn scan(
+        &mut self,
+        range: impl Into<BoundRange>,
+        limit: u32,
+    ) -> Result<impl Iterator<Item = KvPair>> {
+        self.runtime.block_on(self.inner.scan(range, limit))
+    }
+
+    /// Scan a range, return at most `limit` keys that lying in the range.
+    pub fn scan_keys(
+        &mut self,
+        range: impl Into<BoundRange>,
+        limit: u32,
+    ) -> Result<impl Iterator<Item = Key>> {
+        self.runtime.block_on(self.inner.scan_keys(range, limit))
+    }
+
+    /// Similar to scan, but in the reverse direction.
+    pub fn scan_reverse(
+        &mut self,
+        range: impl Into<BoundRange>,
+        limit: u32,
+    ) -> Result<impl Iterator<Item = KvPair>> {
+        self.runtime.block_on(self.inner.scan_reverse(range, limit))
+    }
+
+    /// Similar to scan_keys, but in the reverse direction.
+    pub fn scan_keys_reverse(
+        &mut self,
+        range: impl Into<BoundRange>,
+        limit: u32,
+    ) -> Result<impl Iterator<Item = Key>> {
+        self.runtime
+            .block_on(self.inner.scan_keys_reverse(range, limit))
+    }
+}