Repository: gluesql/gluesql
Pull request #297 (2021-08-19 05:53:33)

Implement SledStorage transaction timeout support,
`SledStorage` users can set transaction timeout.
Transaction terminates when either `ROLLBACK` | `COMMIT` are executed by users or transaction time limit has exceeded.

`SledStorage` now supports `set_transaction_timeout` method to set transaction timeout in milliseconds.
```rust
let storage = SledStorage:new(filename).unwrap();

storage.set_transaction_timeout(Some(3600 * 1000)); // in milliseconds
```

resolve https://github.com/gluesql/gluesql/issues/276


Patch:
diff --git a/src/ast/function.rs b/src/ast/function.rs
index ad9e5099..1334f9f3 100644
--- a/src/ast/function.rs
+++ b/src/ast/function.rs
@@ -9,7 +9,12 @@ pub enum Function {
     Upper(Expr),
     Left { expr: Expr, size: Expr },
     Right { expr: Expr, size: Expr },
+    Ceil(Expr),
+    Round(Expr),
+    Floor(Expr),
     Trim(Expr),
+    Div { dividend: Expr, divisor: Expr },
+    Mod { dividend: Expr, divisor: Expr },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
diff --git a/src/executor/evaluate/error.rs b/src/executor/evaluate/error.rs
index 4fdef3c7..50b6a704 100644
--- a/src/executor/evaluate/error.rs
+++ b/src/executor/evaluate/error.rs
@@ -16,12 +16,18 @@ pub enum EvaluateError {
     #[error("function requires string value: {0}")]
     FunctionRequiresStringValue(String),
 
-    #[error("function requires string value: {0}")]
+    #[error("function requires integer value: {0}")]
     FunctionRequiresIntegerValue(String),
 
+    #[error("function requires float or integer value: {0}")]
+    FunctionRequiresFloatOrIntegerValue(String),
+
     #[error("function requires usize value: {0}")]
     FunctionRequiresUSizeValue(String),
 
+    #[error("function requires float value: {0}")]
+    FunctionRequiresFloatValue(String),
+
     #[error("value not found: {0}")]
     ValueNotFound(String),
 
@@ -42,4 +48,7 @@ pub enum EvaluateError {
 
     #[error("unreachable empty aggregate value: {0:?}")]
     UnreachableEmptyAggregateValue(Aggregate),
+
+    #[error("divisor should not be zero")]
+    InvalidDivisorZero,
 }
diff --git a/src/executor/evaluate/mod.rs b/src/executor/evaluate/mod.rs
index 37f8d687..cb2e63b5 100644
--- a/src/executor/evaluate/mod.rs
+++ b/src/executor/evaluate/mod.rs
@@ -228,6 +228,15 @@ async fn evaluate_function<'a, T: 'static + Debug>(
         }
     };
 
+    let eval_to_float = |name: &'static str, expr| async move {
+        match eval(expr).await?.try_into()? {
+            Value::I64(v) => Ok(Nullable::Value(v as f64)),
+            Value::F64(v) => Ok(Nullable::Value(v)),
+            Value::Null => Ok(Nullable::Null),
+            _ => Err::<_, Error>(EvaluateError::FunctionRequiresFloatValue(name.to_owned()).into()),
+        }
+    };
+
     match func {
         Function::Lower(expr) => match eval_to_str("LOWER", expr).await? {
             Nullable::Value(v) => Ok(Value::Str(v.to_lowercase())),
@@ -281,10 +290,71 @@ async fn evaluate_function<'a, T: 'static + Debug>(
 
             Ok(Evaluated::from(Value::Str(converted)))
         }
+        Function::Ceil(expr) => match eval_to_float("CEIL", expr).await? {
+            Nullable::Value(v) => Ok(Value::F64(v.ceil())),
+            Nullable::Null => Ok(Value::Null),
+        }
+        .map(Evaluated::from),
+        Function::Round(expr) => match eval_to_float("ROUND", expr).await? {
+            Nullable::Value(v) => Ok(Value::F64(v.round())),
+            Nullable::Null => Ok(Value::Null),
+        }
+        .map(Evaluated::from),
+        Function::Floor(expr) => match eval_to_float("FLOOR", expr).await? {
+            Nullable::Value(v) => Ok(Value::F64(v.floor())),
+            Nullable::Null => Ok(Value::Null),
+        }
+        .map(Evaluated::from),
         Function::Trim(expr) => match eval_to_str("TRIM", expr).await? {
             Nullable::Value(string) => Ok(Value::Str(string.trim().to_owned())),
             Nullable::Null => Ok(Value::Null),
         }
         .map(Evaluated::from),
+        Function::Div { dividend, divisor } | Function::Mod { dividend, divisor } => {
+            let name = if matches!(func, Function::Div { .. }) {
+                "DIV"
+            } else {
+                "MOD"
+            };
+
+            let dividend = match eval(dividend).await?.try_into()? {
+                Value::F64(number) => number,
+                Value::I64(number) => number as f64,
+                Value::Null => {
+                    return Ok(Evaluated::from(Value::Null));
+                }
+                _ => {
+                    return Err(EvaluateError::FunctionRequiresFloatOrIntegerValue(
+                        name.to_owned(),
+                    )
+                    .into());
+                }
+            };
+
+            let divisor = match eval(divisor).await?.try_into()? {
+                Value::F64(number) => match number {
+                    x if x == 0.0 => return Err(EvaluateError::InvalidDivisorZero.into()),
+                    _ => number,
+                },
+                Value::I64(number) => match number {
+                    0 => return Err(EvaluateError::InvalidDivisorZero.into()),
+                    _ => number as f64,
+                },
+                Value::Null => {
+                    return Ok(Evaluated::from(Value::Null));
+                }
+                _ => {
+                    return Err(EvaluateError::FunctionRequiresFloatOrIntegerValue(
+                        name.to_owned(),
+                    )
+                    .into());
+                }
+            };
+
+            match name {
+                "DIV" => Ok(Evaluated::from(Value::I64((dividend / divisor) as i64))),
+                _ => Ok(Evaluated::from(Value::F64(dividend % divisor))),
+            }
+        }
     }
 }
diff --git a/src/storages/sled_storage/alter_table.rs b/src/storages/sled_storage/alter_table.rs
index 672f9f2a..7d7c5771 100644
--- a/src/storages/sled_storage/alter_table.rs
+++ b/src/storages/sled_storage/alter_table.rs
@@ -1,13 +1,19 @@
 #![cfg(feature = "alter-table")]
 
 use {
-    super::{error::err_into, error::StorageError, fetch_schema, key, lock, SledStorage, Snapshot},
+    super::{
+        error::err_into,
+        fetch_schema, key,
+        lock::{self, LockAcquired},
+        transaction::TxPayload,
+        SledStorage, Snapshot,
+    },
     crate::{
         ast::ColumnDef, executor::evaluate_stateless, schema::ColumnDefExt, utils::Vector,
         AlterTable, AlterTableError, MutResult, Result, Row, Schema, Value,
     },
     async_trait::async_trait,
-    sled::transaction::{ConflictableTransactionError, TransactionError},
+    sled::transaction::ConflictableTransactionError,
     std::{iter::once, str},
 };
 
@@ -15,27 +21,13 @@ macro_rules! try_self {
     ($self: expr, $expr: expr) => {
         match $expr {
             Err(e) => {
-                return Err(($self, e.into()));
+                return Err(($self, e));
             }
             Ok(v) => v,
         }
     };
 }
 
-macro_rules! transaction {
-    ($self: expr, $expr: expr) => {{
-        let result = $self.tree.transaction($expr).map_err(|e| match e {
-            TransactionError::Abort(e) => e,
-            TransactionError::Storage(e) => StorageError::Sled(e).into(),
-        });
-
-        match result {
-            Ok(_) => Ok(($self, ())),
-            Err(e) => Err(($self, e)),
-        }
-    }};
-}
-
 #[async_trait(?Send)]
 impl AlterTable for SledStorage {
     async fn rename_schema(self, table_name: &str, new_table_name: &str) -> MutResult<Self, ()> {
@@ -48,9 +40,14 @@ impl AlterTable for SledStorage {
         let items = try_self!(self, items);
 
         let state = &self.state;
-
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let (old_schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
             let schema_snapshot = schema_snapshot
@@ -134,8 +131,14 @@ impl AlterTable for SledStorage {
                 tree.insert(temp_new_key, new_schema_key.as_bytes())?;
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.rename_schema(table_name, new_table_name).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn rename_column(
@@ -145,9 +148,14 @@ impl AlterTable for SledStorage {
         new_column_name: &str,
     ) -> MutResult<Self, ()> {
         let state = &self.state;
-
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let (schema_key, snapshot) = fetch_schema(tree, table_name)?;
             let snapshot = snapshot
@@ -197,8 +205,17 @@ impl AlterTable for SledStorage {
                 tree.insert(temp_key, schema_key.as_bytes())?;
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => {
+                self.rename_column(table_name, old_column_name, new_column_name)
+                    .await
+            }
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn add_column(self, table_name: &str, column_def: &ColumnDef) -> MutResult<Self, ()> {
@@ -211,9 +228,14 @@ impl AlterTable for SledStorage {
         let items = try_self!(self, items);
 
         let state = &self.state;
-
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let (schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
             let schema_snapshot = schema_snapshot
@@ -309,8 +331,14 @@ impl AlterTable for SledStorage {
                 tree.insert(temp_key, schema_key.as_bytes())?;
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.add_column(table_name, column_def).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn drop_column(
@@ -328,9 +356,14 @@ impl AlterTable for SledStorage {
         let items = try_self!(self, items);
 
         let state = &self.state;
-
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let (schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
             let schema_snapshot = schema_snapshot
@@ -352,7 +385,7 @@ impl AlterTable for SledStorage {
             let column_index = match (column_index, if_exists) {
                 (Some(index), _) => index,
                 (None, true) => {
-                    return Ok(());
+                    return Ok(TxPayload::Success);
                 }
                 (None, false) => {
                     return Err(
@@ -418,7 +451,13 @@ impl AlterTable for SledStorage {
                 tree.insert(temp_key, schema_key.as_bytes())?;
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.drop_column(table_name, column_name, if_exists).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 }
diff --git a/src/storages/sled_storage/error.rs b/src/storages/sled_storage/error.rs
index 380f77d7..7ac83f3c 100644
--- a/src/storages/sled_storage/error.rs
+++ b/src/storages/sled_storage/error.rs
@@ -1,21 +1,18 @@
-use std::str;
-use thiserror::Error as ThisError;
+use {
+    crate::{Error, IndexError},
+    sled::transaction::TransactionError as SledTransactionError,
+    std::{str, time},
+    thiserror::Error as ThisError,
+};
 
 #[cfg(feature = "alter-table")]
 use crate::AlterTableError;
 
-#[cfg(feature = "index")]
-use crate::IndexError;
-
-use crate::Error;
-
 #[derive(ThisError, Debug)]
 pub enum StorageError {
     #[cfg(feature = "alter-table")]
     #[error(transparent)]
     AlterTable(#[from] AlterTableError),
-
-    #[cfg(feature = "index")]
     #[error(transparent)]
     Index(#[from] IndexError),
 
@@ -25,6 +22,8 @@ pub enum StorageError {
     Bincode(#[from] bincode::Error),
     #[error(transparent)]
     Str(#[from] str::Utf8Error),
+    #[error(transparent)]
+    SystemTime(#[from] time::SystemTimeError),
 }
 
 impl From<StorageError> for Error {
@@ -35,11 +34,10 @@ impl From<StorageError> for Error {
             Sled(e) => Error::Storage(Box::new(e)),
             Bincode(e) => Error::Storage(e),
             Str(e) => Error::Storage(Box::new(e)),
+            SystemTime(e) => Error::Storage(Box::new(e)),
 
             #[cfg(feature = "alter-table")]
             AlterTable(e) => e.into(),
-
-            #[cfg(feature = "index")]
             Index(e) => e.into(),
         }
     }
@@ -54,3 +52,10 @@ where
 
     e
 }
+
+pub fn tx_err_into(e: SledTransactionError<Error>) -> Error {
+    match e {
+        SledTransactionError::Abort(e) => e,
+        SledTransactionError::Storage(e) => StorageError::Sled(e).into(),
+    }
+}
diff --git a/src/storages/sled_storage/gc.rs b/src/storages/sled_storage/gc.rs
index ef21abe7..0de093a4 100644
--- a/src/storages/sled_storage/gc.rs
+++ b/src/storages/sled_storage/gc.rs
@@ -8,10 +8,25 @@ use {
         data::{Row, Schema},
         result::Result,
     },
+    std::time::{SystemTime, UNIX_EPOCH},
 };
 
 impl SledStorage {
     pub fn gc(&self) -> Result<()> {
+        let mut lock: Lock = self
+            .tree
+            .get("lock/")
+            .map_err(err_into)?
+            .map(|l| bincode::deserialize(&l))
+            .transpose()
+            .map_err(err_into)?
+            .unwrap_or_default();
+
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(err_into)?
+            .as_millis();
+
         let txids = self
             .tree
             .scan_prefix("tx_data/")
@@ -20,9 +35,19 @@ impl SledStorage {
                     .map_err(err_into)?
                     .map_err(err_into)
             })
-            .take_while(|tx_data| match tx_data {
-                Ok(TxData { alive, .. }) => !alive,
-                Err(_) => false,
+            .take_while(|tx_data| match (tx_data, self.tx_timeout) {
+                (Ok(TxData { alive, .. }), None) => !alive,
+                (Ok(tx_data), Some(tx_timeout)) => {
+                    let TxData {
+                        txid,
+                        alive,
+                        created_at,
+                    } = tx_data;
+
+                    (!alive || now - created_at >= tx_timeout)
+                        && Some(txid) != lock.lock_txid.as_ref()
+                }
+                (Err(_), _) => false,
             })
             .map(|tx_data| tx_data.map(|TxData { txid, .. }| txid))
             .collect::<Result<Vec<u64>>>()?;
@@ -34,19 +59,7 @@ impl SledStorage {
             }
         };
 
-        let Lock { lock_txid, .. } = self
-            .tree
-            .get("lock/")
-            .map_err(err_into)?
-            .map(|l| bincode::deserialize(&l))
-            .transpose()
-            .map_err(err_into)?
-            .unwrap_or_default();
-
-        let lock = Lock {
-            lock_txid,
-            gc_txid: Some(*max_txid),
-        };
+        lock.gc_txid = Some(*max_txid);
 
         bincode::serialize(&lock)
             .map(|lock| self.tree.insert("lock/", lock))
diff --git a/src/storages/sled_storage/index.rs b/src/storages/sled_storage/index.rs
index 36060fc7..12e68017 100644
--- a/src/storages/sled_storage/index.rs
+++ b/src/storages/sled_storage/index.rs
@@ -83,15 +83,17 @@ impl Index<IVec> for SledStorage {
             }
         };
 
-        let txid = match self.state {
-            State::Transaction { txid, .. } => txid,
+        let (txid, created_at) = match self.state {
+            State::Transaction {
+                txid, created_at, ..
+            } => (txid, created_at),
             State::Idle => {
                 return Err(Error::StorageMsg(
                     "conflict - scan_indexed_data failed, lock does not exist".to_owned(),
                 ));
             }
         };
-        let lock_txid = lock::fetch(&self.tree, txid)?;
+        let lock_txid = lock::fetch(&self.tree, txid, created_at, self.tx_timeout)?;
 
         let tree = self.tree.clone();
         let flat_map = move |keys: Result<IVec>| {
diff --git a/src/storages/sled_storage/index_mut.rs b/src/storages/sled_storage/index_mut.rs
index 37c65317..2446db2d 100644
--- a/src/storages/sled_storage/index_mut.rs
+++ b/src/storages/sled_storage/index_mut.rs
@@ -1,6 +1,11 @@
 use {
     super::{
-        err_into, error::StorageError, index_sync::IndexSync, key, lock, SledStorage, Snapshot,
+        err_into,
+        index_sync::IndexSync,
+        key,
+        lock::{self, LockAcquired},
+        transaction::TxPayload,
+        SledStorage, Snapshot,
     },
     crate::{
         ast::OrderByExpr,
@@ -12,8 +17,7 @@ use {
     async_trait::async_trait,
     sled::{
         transaction::{
-            ConflictableTransactionError, ConflictableTransactionResult, TransactionError,
-            TransactionalTree,
+            ConflictableTransactionError, ConflictableTransactionResult, TransactionalTree,
         },
         IVec,
     },
@@ -31,20 +35,6 @@ macro_rules! try_self {
     };
 }
 
-macro_rules! transaction {
-    ($self: expr, $expr: expr) => {{
-        let result = $self.tree.transaction($expr).map_err(|e| match e {
-            TransactionError::Abort(e) => e,
-            TransactionError::Storage(e) => StorageError::Sled(e).into(),
-        });
-
-        match result {
-            Ok(_) => Ok(($self, ())),
-            Err(e) => Err(($self, e)),
-        }
-    }};
-}
-
 fn fetch_schema(
     tree: &TransactionalTree,
     table_name: &str,
@@ -72,9 +62,14 @@ impl IndexMut<IVec> for SledStorage {
         let rows = try_self!(self, rows.collect::<Result<Vec<_>>>());
 
         let state = &self.state;
-
-        transaction!(self, |tree| {
-            let (txid, _) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let txid = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, .. } => txid,
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let index_expr = &column.expr;
 
@@ -130,8 +125,14 @@ impl IndexMut<IVec> for SledStorage {
             let temp_key = key::temp_schema(txid, table_name);
             tree.insert(temp_key, schema_key.as_bytes())?;
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.create_index(table_name, index_name, column).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn drop_index(self, table_name: &str, index_name: &str) -> MutResult<Self, ()> {
@@ -139,9 +140,14 @@ impl IndexMut<IVec> for SledStorage {
         let rows = try_self!(self, rows.collect::<Result<Vec<_>>>());
 
         let state = &self.state;
-
-        transaction!(self, |tree| {
-            let (txid, _) = lock::acquire(tree, state)?;
+        let tx_timeout = self.tx_timeout;
+        let tx_result = self.tree.transaction(move |tree| {
+            let txid = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, .. } => txid,
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let (schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
             let schema_snapshot = schema_snapshot
@@ -191,7 +197,13 @@ impl IndexMut<IVec> for SledStorage {
             let temp_key = key::temp_schema(txid, table_name);
             tree.insert(temp_key, schema_key.as_bytes())?;
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.drop_index(table_name, index_name).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 }
diff --git a/src/storages/sled_storage/lock.rs b/src/storages/sled_storage/lock.rs
index 7c3531ca..c8148100 100644
--- a/src/storages/sled_storage/lock.rs
+++ b/src/storages/sled_storage/lock.rs
@@ -8,19 +8,20 @@ use {
         },
         Db,
     },
+    std::time::{SystemTime, UNIX_EPOCH},
 };
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct TxData {
     pub txid: u64,
     pub alive: bool,
-    // TODO: support timeout based garbage collection
-    // - created_at: u128,
+    pub created_at: u128,
 }
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct Lock {
     pub lock_txid: Option<u64>,
+    pub lock_created_at: u128,
     pub gc_txid: Option<u64>,
     // TODO: support serializable transaction isolation level
     // - prev_done_at: u128,
@@ -35,21 +36,37 @@ pub fn get_txdata_key(txid: u64) -> Vec<u8> {
         .collect::<Vec<_>>()
 }
 
-pub fn register(tree: &Db) -> Result<u64> {
+pub fn register(tree: &Db) -> Result<(u64, u128)> {
     let txid = tree.generate_id().map_err(err_into)?;
     let key = get_txdata_key(txid);
-    let tx_data = TxData { txid, alive: true };
+    let created_at = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(err_into)?
+        .as_millis();
+
+    let tx_data = TxData {
+        txid,
+        alive: true,
+        created_at,
+    };
 
     bincode::serialize(&tx_data)
         .map_err(err_into)
         .map(|tx_data| tree.insert(key, tx_data))?
         .map_err(err_into)?;
 
-    Ok(txid)
+    Ok((txid, created_at))
 }
 
-pub fn fetch(tree: &Db, txid: u64) -> Result<Option<u64>> {
-    let Lock { lock_txid, gc_txid } = tree
+pub fn fetch(
+    tree: &Db,
+    txid: u64,
+    created_at: u128,
+    tx_timeout: Option<u128>,
+) -> Result<Option<u64>> {
+    let Lock {
+        lock_txid, gc_txid, ..
+    } = tree
         .get("lock/")
         .map_err(err_into)?
         .map(|l| bincode::deserialize(&l))
@@ -57,20 +74,39 @@ pub fn fetch(tree: &Db, txid: u64) -> Result<Option<u64>> {
         .map_err(err_into)?
         .unwrap_or_default();
 
-    if gc_txid.is_some() && Some(txid) <= gc_txid {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(err_into)?
+        .as_millis();
+
+    if tx_timeout.map(|tx_timeout| now - created_at >= tx_timeout) == Some(true) {
+        return Err(Error::StorageMsg(
+            "fetch failed - expired transaction has used (timeout)".to_owned(),
+        ));
+    } else if gc_txid.is_some() && Some(txid) <= gc_txid {
         return Err(Error::StorageMsg(
-            "fetch failed - expired transaction is used".to_owned(),
+            "fetch failed - expired transaction has used (txid)".to_owned(),
         ));
     }
 
     Ok(lock_txid)
 }
 
+pub enum LockAcquired {
+    Success { txid: u64, autocommit: bool },
+    RollbackAndRetry { lock_txid: u64 },
+}
+
 pub fn acquire(
     tree: &TransactionalTree,
     state: &State,
-) -> ConflictableTransactionResult<(u64, bool), Error> {
-    let Lock { lock_txid, gc_txid } = tree
+    tx_timeout: Option<u128>,
+) -> ConflictableTransactionResult<LockAcquired, Error> {
+    let Lock {
+        lock_txid,
+        lock_created_at,
+        gc_txid,
+    } = tree
         .get("lock/")?
         .map(|l| bincode::deserialize(&l))
         .transpose()
@@ -78,8 +114,12 @@ pub fn acquire(
         .map_err(ConflictableTransactionError::Abort)?
         .unwrap_or_default();
 
-    let (txid, autocommit) = match state {
-        State::Transaction { txid, autocommit } => (txid, autocommit),
+    let (txid, created_at, autocommit) = match state {
+        State::Transaction {
+            txid,
+            created_at,
+            autocommit,
+        } => (*txid, *created_at, *autocommit),
         State::Idle => {
             return Err(Error::StorageMsg(
                 "conflict - cannot acquire lock from idle state".to_owned(),
@@ -88,22 +128,39 @@ pub fn acquire(
         }
     };
 
-    if gc_txid.is_some() && Some(txid) <= gc_txid.as_ref() {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(err_into)
+        .map_err(ConflictableTransactionError::Abort)?
+        .as_millis();
+
+    if tx_timeout.map(|tx_timeout| now - created_at >= tx_timeout) == Some(true) {
+        return Err(Error::StorageMsg(
+            "acquire failed - expired transaction has used (timeout)".to_owned(),
+        ))
+        .map_err(ConflictableTransactionError::Abort);
+    } else if gc_txid.is_some() && Some(txid) <= gc_txid {
         return Err(Error::StorageMsg(
-            "acquire failed - expired transaction is used".to_owned(),
+            "acquire failed - expired transaction has used (txid)".to_owned(),
         ))
         .map_err(ConflictableTransactionError::Abort);
     }
 
     let txid = match lock_txid {
-        Some(lock_txid) if txid == &lock_txid => txid,
-        Some(_) => {
-            return Err(Error::StorageMsg("database is locked".to_owned()))
-                .map_err(ConflictableTransactionError::Abort);
+        Some(lock_txid) => {
+            if tx_timeout.map(|tx_timeout| now - lock_created_at >= tx_timeout) == Some(true) {
+                return Ok(LockAcquired::RollbackAndRetry { lock_txid });
+            } else if txid != lock_txid {
+                return Err(Error::StorageMsg("database is locked".to_owned()))
+                    .map_err(ConflictableTransactionError::Abort);
+            }
+
+            txid
         }
         None => {
             let lock = Lock {
-                lock_txid: Some(*txid),
+                lock_txid: Some(txid),
+                lock_created_at: created_at,
                 gc_txid,
             };
 
@@ -116,7 +173,7 @@ pub fn acquire(
         }
     };
 
-    Ok((*txid, *autocommit))
+    Ok(LockAcquired::Success { txid, autocommit })
 }
 
 pub fn unregister(tree: &Db, txid: u64) -> Result<()> {
@@ -139,7 +196,9 @@ pub fn unregister(tree: &Db, txid: u64) -> Result<()> {
 }
 
 pub fn release(tree: &TransactionalTree, txid: u64) -> ConflictableTransactionResult<(), Error> {
-    let Lock { gc_txid, .. } = tree
+    let Lock {
+        gc_txid, lock_txid, ..
+    } = tree
         .get("lock/")?
         .map(|l| bincode::deserialize(&l))
         .transpose()
@@ -147,25 +206,34 @@ pub fn release(tree: &TransactionalTree, txid: u64) -> ConflictableTransactionRe
         .map_err(ConflictableTransactionError::Abort)?
         .unwrap_or_default();
 
-    let lock = Lock {
-        lock_txid: None,
-        gc_txid,
-    };
-
-    bincode::serialize(&lock)
-        .map_err(err_into)
-        .map_err(ConflictableTransactionError::Abort)
-        .map(|lock| tree.insert("lock/", lock))??;
+    if Some(txid) == lock_txid {
+        let lock = Lock {
+            lock_txid: None,
+            lock_created_at: 0,
+            gc_txid,
+        };
+
+        bincode::serialize(&lock)
+            .map_err(err_into)
+            .map_err(ConflictableTransactionError::Abort)
+            .map(|lock| tree.insert("lock/", lock))??;
+    }
 
     let key = get_txdata_key(txid);
-    let mut tx_data: TxData = tree
+    let tx_data: Option<TxData> = tree
         .get(&key)?
-        .ok_or_else(|| Error::StorageMsg("conflict - lock does not exist".to_owned()))
         .map(|tx_data| bincode::deserialize(&tx_data))
-        .map_err(ConflictableTransactionError::Abort)?
+        .transpose()
         .map_err(err_into)
         .map_err(ConflictableTransactionError::Abort)?;
 
+    let mut tx_data = match tx_data {
+        Some(tx_data) => tx_data,
+        None => {
+            return Ok(());
+        }
+    };
+
     tx_data.alive = false;
 
     bincode::serialize(&tx_data)
diff --git a/src/storages/sled_storage/mod.rs b/src/storages/sled_storage/mod.rs
index de9820a2..603bef75 100644
--- a/src/storages/sled_storage/mod.rs
+++ b/src/storages/sled_storage/mod.rs
@@ -14,10 +14,11 @@ mod transaction;
 use {
     self::snapshot::Snapshot,
     crate::{
+        data::Schema,
         store::{GStore, GStoreMut},
         Error, Result,
     },
-    error::err_into,
+    error::{err_into, tx_err_into},
     sled::{
         self,
         transaction::{
@@ -28,24 +29,50 @@ use {
     std::convert::TryFrom,
 };
 
+/// default transaction timeout : 1 hour
+const DEFAULT_TX_TIMEOUT: u128 = 3600 * 1000;
+
 #[derive(Debug, Clone)]
 pub enum State {
     Idle,
-    Transaction { txid: u64, autocommit: bool },
+    Transaction {
+        txid: u64,
+        created_at: u128,
+        autocommit: bool,
+    },
 }
 
 #[derive(Debug, Clone)]
 pub struct SledStorage {
     pub tree: Db,
     pub state: State,
+    /// transaction timeout in milliseconds
+    pub tx_timeout: Option<u128>,
 }
 
 impl SledStorage {
     pub fn new(filename: &str) -> Result<Self> {
         let tree = sled::open(filename).map_err(err_into)?;
         let state = State::Idle;
+        let tx_timeout = Some(DEFAULT_TX_TIMEOUT);
+
+        Ok(Self {
+            tree,
+            state,
+            tx_timeout,
+        })
+    }
+
+    pub fn set_transaction_timeout(&mut self, tx_timeout: Option<u128>) {
+        self.tx_timeout = tx_timeout;
+    }
 
-        Ok(Self { tree, state })
+    fn update_state(self, state: State) -> Self {
+        Self {
+            tree: self.tree,
+            state,
+            tx_timeout: self.tx_timeout,
+        }
     }
 }
 
@@ -55,15 +82,20 @@ impl TryFrom<Config> for SledStorage {
     fn try_from(config: Config) -> Result<Self> {
         let tree = config.open().map_err(err_into)?;
         let state = State::Idle;
+        let tx_timeout = Some(DEFAULT_TX_TIMEOUT);
 
-        Ok(Self { tree, state })
+        Ok(Self {
+            tree,
+            state,
+            tx_timeout,
+        })
     }
 }
 
 fn fetch_schema(
     tree: &TransactionalTree,
     table_name: &str,
-) -> ConflictableTransactionResult<(String, Option<Snapshot<crate::data::Schema>>), Error> {
+) -> ConflictableTransactionResult<(String, Option<Snapshot<Schema>>), Error> {
     let key = format!("schema/{}", table_name);
     let value = tree.get(&key.as_bytes())?;
     let schema_snapshot = value
diff --git a/src/storages/sled_storage/store.rs b/src/storages/sled_storage/store.rs
index 4d88e504..650e6bbc 100644
--- a/src/storages/sled_storage/store.rs
+++ b/src/storages/sled_storage/store.rs
@@ -12,11 +12,15 @@ use {
 #[async_trait(?Send)]
 impl Store<IVec> for SledStorage {
     async fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>> {
-        let (txid, temp) = match self.state {
-            State::Transaction { txid, .. } => (txid, false),
-            State::Idle => (lock::register(&self.tree)?, true),
+        let (txid, created_at, temp) = match self.state {
+            State::Transaction {
+                txid, created_at, ..
+            } => (txid, created_at, false),
+            State::Idle => {
+                lock::register(&self.tree).map(|(txid, created_at)| (txid, created_at, true))?
+            }
         };
-        let lock_txid = lock::fetch(&self.tree, txid)?;
+        let lock_txid = lock::fetch(&self.tree, txid, created_at, self.tx_timeout)?;
 
         let key = format!("schema/{}", table_name);
         let schema = self
@@ -37,15 +41,17 @@ impl Store<IVec> for SledStorage {
     }
 
     async fn scan_data(&self, table_name: &str) -> Result<RowIter<IVec>> {
-        let txid = match self.state {
-            State::Transaction { txid, .. } => txid,
+        let (txid, created_at) = match self.state {
+            State::Transaction {
+                txid, created_at, ..
+            } => (txid, created_at),
             State::Idle => {
                 return Err(Error::StorageMsg(
                     "conflict - scan_data failed, lock does not exist".to_owned(),
                 ));
             }
         };
-        let lock_txid = lock::fetch(&self.tree, txid)?;
+        let lock_txid = lock::fetch(&self.tree, txid, created_at, self.tx_timeout)?;
 
         let prefix = format!("data/{}/", table_name);
         let result_set = self
diff --git a/src/storages/sled_storage/store_mut.rs b/src/storages/sled_storage/store_mut.rs
index 26c11403..e4691b74 100644
--- a/src/storages/sled_storage/store_mut.rs
+++ b/src/storages/sled_storage/store_mut.rs
@@ -1,6 +1,11 @@
 use {
     super::{
-        err_into, error::StorageError, index_sync::IndexSync, key, lock, SledStorage, Snapshot,
+        err_into,
+        index_sync::IndexSync,
+        key,
+        lock::{self, LockAcquired},
+        transaction::TxPayload,
+        SledStorage, Snapshot,
     },
     crate::{
         data::{Row, Schema},
@@ -8,33 +13,23 @@ use {
         IndexError, MutResult, StoreMut,
     },
     async_trait::async_trait,
-    sled::{
-        transaction::{ConflictableTransactionError, TransactionError},
-        IVec,
-    },
+    sled::{transaction::ConflictableTransactionError, IVec},
+    std::rc::Rc,
 };
 
-macro_rules! transaction {
-    ($self: expr, $expr: expr) => {{
-        let result = $self.tree.transaction($expr).map_err(|e| match e {
-            TransactionError::Abort(e) => e,
-            TransactionError::Storage(e) => StorageError::Sled(e).into(),
-        });
-
-        match result {
-            Ok(_) => Ok(($self, ())),
-            Err(e) => Err(($self, e)),
-        }
-    }};
-}
-
 #[async_trait(?Send)]
 impl StoreMut<IVec> for SledStorage {
     async fn insert_schema(self, schema: &Schema) -> MutResult<Self, ()> {
         let state = &self.state;
+        let tx_timeout = self.tx_timeout;
 
-        transaction!(self, move |tree| {
-            let (txid, _) = lock::acquire(tree, state)?;
+        let tx_result = self.tree.transaction(move |tree| {
+            let txid = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, .. } => txid,
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let key = format!("schema/{}", schema.table_name);
             let temp_key = key::temp_schema(txid, &schema.table_name);
@@ -58,8 +53,14 @@ impl StoreMut<IVec> for SledStorage {
             tree.insert(key.as_bytes(), snapshot)?;
             tree.insert(temp_key, key.as_bytes())?;
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.insert_schema(schema).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn delete_schema(self, table_name: &str) -> MutResult<Self, ()> {
@@ -77,9 +78,15 @@ impl StoreMut<IVec> for SledStorage {
         };
 
         let state = &self.state;
+        let tx_timeout = self.tx_timeout;
 
-        transaction!(self, move |tree| {
-            let (txid, _) = lock::acquire(tree, state)?;
+        let tx_result = self.tree.transaction(move |tree| {
+            let txid = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, .. } => txid,
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
             let key = format!("schema/{}", table_name);
             let temp_key = key::temp_schema(txid, table_name);
@@ -94,7 +101,7 @@ impl StoreMut<IVec> for SledStorage {
             let (snapshot, schema) = match snapshot.map(|snapshot| snapshot.delete(txid)) {
                 Some((snapshot, Some(schema))) => (snapshot, schema),
                 Some((_, None)) | None => {
-                    return Ok(());
+                    return Ok(TxPayload::Success);
                 }
             };
             let snapshot = bincode::serialize(&snapshot)
@@ -132,18 +139,45 @@ impl StoreMut<IVec> for SledStorage {
                 index_sync.delete(row_key, &deleted_row)?;
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.delete_schema(table_name).await,
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
     async fn insert_data(self, table_name: &str, rows: Vec<Row>) -> MutResult<Self, ()> {
+        self.insert_data_sync(table_name, Rc::new(rows))
+    }
+
+    async fn update_data(self, table_name: &str, rows: Vec<(IVec, Row)>) -> MutResult<Self, ()> {
+        self.update_data_sync(table_name, Rc::new(rows))
+    }
+
+    async fn delete_data(self, table_name: &str, keys: Vec<IVec>) -> MutResult<Self, ()> {
+        self.delete_data_sync(table_name, Rc::new(keys))
+    }
+}
+
+impl SledStorage {
+    fn insert_data_sync(self, table_name: &str, rows: Rc<Vec<Row>>) -> MutResult<Self, ()> {
         let state = &self.state;
+        let tx_timeout = self.tx_timeout;
+        let tx_rows = Rc::clone(&rows);
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
             let index_sync = IndexSync::new(tree, txid, table_name)?;
 
-            for row in rows.iter() {
+            for row in tx_rows.iter() {
                 let id = tree.generate_id()?;
                 let id = id.to_be_bytes();
                 let prefix = format!("data/{}/", table_name);
@@ -172,18 +206,31 @@ impl StoreMut<IVec> for SledStorage {
                 }
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.insert_data_sync(table_name, rows),
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
-    async fn update_data(self, table_name: &str, rows: Vec<(IVec, Row)>) -> MutResult<Self, ()> {
+    fn update_data_sync(self, table_name: &str, rows: Rc<Vec<(IVec, Row)>>) -> MutResult<Self, ()> {
         let state = &self.state;
+        let tx_timeout = self.tx_timeout;
+        let tx_rows = Rc::clone(&rows);
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
-        transaction!(self, move |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
             let index_sync = IndexSync::new(tree, txid, table_name)?;
 
-            for (key, new_row) in rows.iter() {
+            for (key, new_row) in tx_rows.iter() {
                 let snapshot = tree
                     .get(key)?
                     .ok_or_else(|| IndexError::ConflictOnEmptyIndexValueDelete.into())
@@ -214,18 +261,31 @@ impl StoreMut<IVec> for SledStorage {
                 }
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.update_data_sync(table_name, rows),
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 
-    async fn delete_data(self, table_name: &str, keys: Vec<IVec>) -> MutResult<Self, ()> {
+    fn delete_data_sync(self, table_name: &str, keys: Rc<Vec<IVec>>) -> MutResult<Self, ()> {
         let state = &self.state;
+        let tx_timeout = self.tx_timeout;
+        let tx_keys = Rc::clone(&keys);
+        let tx_result = self.tree.transaction(move |tree| {
+            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
+                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
+                LockAcquired::RollbackAndRetry { lock_txid } => {
+                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
+                }
+            };
 
-        transaction!(self, |tree| {
-            let (txid, autocommit) = lock::acquire(tree, state)?;
             let index_sync = IndexSync::new(tree, txid, table_name)?;
 
-            for key in keys.iter() {
+            for key in tx_keys.iter() {
                 let snapshot = tree
                     .get(key)?
                     .ok_or_else(|| IndexError::ConflictOnEmptyIndexValueDelete.into())
@@ -256,7 +316,13 @@ impl StoreMut<IVec> for SledStorage {
                 }
             }
 
-            Ok(())
-        })
+            Ok(TxPayload::Success)
+        });
+
+        match self.check_retry(tx_result) {
+            Ok(true) => self.delete_data_sync(table_name, keys),
+            Ok(false) => Ok((self, ())),
+            Err(e) => Err((self, e)),
+        }
     }
 }
diff --git a/src/storages/sled_storage/transaction.rs b/src/storages/sled_storage/transaction.rs
index 3973e3ba..56d6f7fa 100644
--- a/src/storages/sled_storage/transaction.rs
+++ b/src/storages/sled_storage/transaction.rs
@@ -1,28 +1,27 @@
-#![cfg(feature = "transaction")]
-
 use {
-    super::{err_into, error::StorageError, key, lock, SledStorage, Snapshot, State},
+    super::{
+        err_into,
+        error::StorageError,
+        key,
+        lock::{self, Lock},
+        tx_err_into, SledStorage, Snapshot, State,
+    },
     crate::{
         data::{Row, Schema},
         Error, MutResult, Result, Transaction,
     },
     async_trait::async_trait,
-    sled::transaction::{ConflictableTransactionError, TransactionError},
+    serde::{de::DeserializeOwned, Serialize},
+    sled::{
+        transaction::{
+            ConflictableTransactionError, ConflictableTransactionResult, TransactionError,
+            TransactionalTree,
+        },
+        IVec,
+    },
+    std::{fmt::Debug, result::Result as StdResult},
 };
 
-macro_rules! try_block {
-    ($self: expr, $block: block) => {{
-        let block = || $block;
-
-        match block() {
-            Err(e) => {
-                return Err(($self, e));
-            }
-            Ok(v) => v,
-        }
-    }};
-}
-
 macro_rules! transaction {
     ($self: expr, $expr: expr) => {{
         let result = $self.tree.transaction($expr).map_err(|e| match e {
@@ -31,38 +30,51 @@ macro_rules! transaction {
         });
 
         match result {
-            Ok(_) => {
+            Ok(v) => {
                 let storage = Self {
                     tree: $self.tree,
                     state: State::Idle,
+                    tx_timeout: $self.tx_timeout,
                 };
 
-                Ok((storage, ()))
+                Ok((storage, v))
             }
             Err(e) => Err(($self, e)),
         }
     }};
 }
 
+pub enum TxPayload {
+    Success,
+    RollbackAndRetry(u64),
+}
+
 #[async_trait(?Send)]
 impl Transaction for SledStorage {
     async fn begin(self, autocommit: bool) -> MutResult<Self, bool> {
-        let (txid, autocommit) = try_block!(self, {
-            match (&self.state, autocommit) {
-                (State::Transaction { .. }, false) => Err(Error::StorageMsg(
-                    "nested transaction is not supported".to_owned(),
-                )),
-                (State::Transaction { txid, autocommit }, true) => Ok((*txid, *autocommit)),
-                (State::Idle, _) => lock::register(&self.tree).map(|txid| (txid, autocommit)),
+        match (&self.state, autocommit) {
+            (State::Transaction { .. }, false) => Err((
+                self,
+                Error::StorageMsg("nested transaction is not supported".to_owned()),
+            )),
+            (State::Transaction { autocommit, .. }, true) => {
+                let autocommit = *autocommit;
+
+                Ok((self, autocommit))
             }
-        });
-
-        let storage = Self {
-            tree: self.tree,
-            state: State::Transaction { txid, autocommit },
-        };
-
-        Ok((storage, autocommit))
+            (State::Idle, _) => match lock::register(&self.tree) {
+                Ok((txid, created_at)) => {
+                    let state = State::Transaction {
+                        txid,
+                        created_at,
+                        autocommit,
+                    };
+
+                    Ok((self.update_state(state), autocommit))
+                }
+                Err(e) => Err((self, e)),
+            },
+        }
     }
 
     async fn rollback(self) -> MutResult<Self, ()> {
@@ -76,73 +88,88 @@ impl Transaction for SledStorage {
             }
         };
 
-        let fetch_items = |prefix| {
-            self.tree
-                .scan_prefix(prefix)
-                .map(|item| item.map_err(err_into))
-                .collect::<Result<Vec<_>>>()
+        let rollback = || {
+            let Lock { lock_txid, .. } = self
+                .tree
+                .get("lock/")
+                .map_err(err_into)?
+                .map(|l| bincode::deserialize(&l))
+                .transpose()
+                .map_err(err_into)?
+                .unwrap_or_default();
+
+            if Some(txid) == lock_txid {
+                self.rollback_txid(txid).map(|_| lock_txid)
+            } else {
+                Ok(None)
+            }
         };
 
-        let temp_items = try_block!(self, {
-            let lock_txid = lock::fetch(&self.tree, txid)?;
+        match rollback() {
+            Ok(lock_txid) => transaction!(self, move |tree| {
+                lock_txid
+                    .map(|lock_txid| lock::release(tree, lock_txid))
+                    .transpose()
+            })
+            .map(|(storage, _)| (storage.update_state(State::Idle), ())),
+            Err(e) => Err((self, e)),
+        }
+    }
 
-            if Some(txid) != lock_txid {
-                return Ok(None);
+    async fn commit(self) -> MutResult<Self, ()> {
+        let (txid, created_at) = match self.state {
+            State::Transaction {
+                txid, created_at, ..
+            } => (txid, created_at),
+            State::Idle => {
+                return Err((
+                    self,
+                    Error::StorageMsg("no transaction to commit".to_owned()),
+                ));
             }
+        };
 
-            let data_items = fetch_items(key::temp_data_prefix(txid))?;
-            let schema_items = fetch_items(key::temp_schema_prefix(txid))?;
-            let index_items = fetch_items(key::temp_index_prefix(txid))?;
-
-            Ok(Some((data_items, schema_items, index_items)))
-        });
+        if let Err(e) = lock::fetch(&self.tree, txid, created_at, self.tx_timeout) {
+            return Err((self, e));
+        }
 
-        let (data_items, schema_items, index_items) = match temp_items {
-            Some(items) => items,
-            None => {
-                return Ok((
-                    Self {
-                        tree: self.tree,
-                        state: State::Idle,
-                    },
-                    (),
-                ));
+        let (storage, _) = transaction!(self, move |tree| { lock::release(tree, txid) })?;
+        let gc = || {
+            if storage.tree.get("gc_lock").map_err(err_into)?.is_some() {
+                return Ok(());
             }
-        };
 
-        transaction!(self, move |tree| {
-            for (temp_key, value_key) in data_items.iter() {
-                tree.remove(temp_key)?;
+            storage.tree.insert("gc_lock", &[1]).map_err(err_into)?;
 
-                let snapshot = tree
-                    .get(value_key)?
-                    .map(|l| bincode::deserialize(&l))
-                    .transpose()
-                    .map_err(err_into)
-                    .map_err(ConflictableTransactionError::Abort)?;
+            let gc_result = storage.gc();
 
-                let snapshot: Snapshot<Row> = match snapshot {
-                    Some(snapshot) => snapshot,
-                    None => {
-                        continue;
-                    }
-                };
+            storage.tree.remove("gc_lock").map_err(err_into)?;
 
-                match snapshot.rollback(txid) {
-                    Some(snapshot) => {
-                        let snapshot = bincode::serialize(&snapshot)
-                            .map_err(err_into)
-                            .map_err(ConflictableTransactionError::Abort)?;
+            gc_result
+        };
 
-                        tree.insert(value_key, snapshot)?;
-                    }
-                    None => {
-                        tree.remove(value_key)?;
-                    }
-                };
-            }
+        match gc() {
+            Ok(_) => Ok((storage, ())),
+            Err(e) => Err((storage, e)),
+        }
+    }
+}
+
+impl SledStorage {
+    pub fn rollback_txid(&self, txid: u64) -> Result<()> {
+        let fetch_items = |prefix| {
+            self.tree
+                .scan_prefix(prefix)
+                .map(|item| item.map_err(err_into))
+                .collect::<Result<Vec<_>>>()
+        };
 
-            for (temp_key, value_key) in schema_items.iter() {
+        fn rollback_items<T: Debug + Clone + Serialize + DeserializeOwned>(
+            tree: &TransactionalTree,
+            txid: u64,
+            items: &[(IVec, IVec)],
+        ) -> ConflictableTransactionResult<(), Error> {
+            for (temp_key, value_key) in items.iter() {
                 tree.remove(temp_key)?;
 
                 let snapshot = tree
@@ -152,7 +179,7 @@ impl Transaction for SledStorage {
                     .map_err(err_into)
                     .map_err(ConflictableTransactionError::Abort)?;
 
-                let snapshot: Snapshot<Schema> = match snapshot {
+                let snapshot: Snapshot<T> = match snapshot {
                     Some(snapshot) => snapshot,
                     None => {
                         continue;
@@ -173,76 +200,71 @@ impl Transaction for SledStorage {
                 };
             }
 
-            for (temp_key, value_key) in index_items.iter() {
-                tree.remove(temp_key)?;
+            Ok(())
+        }
 
-                let snapshots = tree
-                    .get(value_key)?
-                    .map(|l| bincode::deserialize(&l))
-                    .transpose()
-                    .map_err(err_into)
-                    .map_err(ConflictableTransactionError::Abort)?;
+        let data_items = fetch_items(key::temp_data_prefix(txid))?;
+        let schema_items = fetch_items(key::temp_schema_prefix(txid))?;
+        let index_items = fetch_items(key::temp_index_prefix(txid))?;
 
-                let snapshots: Vec<Snapshot<Vec<u8>>> = match snapshots {
-                    Some(snapshots) => snapshots,
-                    None => {
-                        continue;
-                    }
-                };
+        self.tree
+            .transaction(move |tree| {
+                rollback_items::<Row>(tree, txid, &data_items)?;
+                rollback_items::<Schema>(tree, txid, &schema_items)?;
 
-                let snapshots = snapshots
-                    .into_iter()
-                    .filter_map(|snapshot| snapshot.rollback(txid))
-                    .collect::<Vec<_>>();
+                for (temp_key, value_key) in index_items.iter() {
+                    tree.remove(temp_key)?;
 
-                if snapshots.is_empty() {
-                    tree.remove(value_key)?;
-                } else {
-                    let snapshots = bincode::serialize(&snapshots)
+                    let snapshots = tree
+                        .get(value_key)?
+                        .map(|l| bincode::deserialize(&l))
+                        .transpose()
                         .map_err(err_into)
                         .map_err(ConflictableTransactionError::Abort)?;
 
-                    tree.insert(value_key, snapshots)?;
-                }
-            }
+                    let snapshots: Vec<Snapshot<Vec<u8>>> = match snapshots {
+                        Some(snapshots) => snapshots,
+                        None => {
+                            continue;
+                        }
+                    };
 
-            lock::release(tree, txid)?;
+                    let snapshots = snapshots
+                        .into_iter()
+                        .filter_map(|snapshot| snapshot.rollback(txid))
+                        .collect::<Vec<_>>();
 
-            Ok(())
-        })
-    }
-
-    async fn commit(self) -> MutResult<Self, ()> {
-        let txid = match self.state {
-            State::Transaction { txid, .. } => txid,
-            State::Idle => {
-                return Err((
-                    self,
-                    Error::StorageMsg("no transaction to commit".to_owned()),
-                ));
-            }
-        };
+                    if snapshots.is_empty() {
+                        tree.remove(value_key)?;
+                    } else {
+                        let snapshots = bincode::serialize(&snapshots)
+                            .map_err(err_into)
+                            .map_err(ConflictableTransactionError::Abort)?;
 
-        let (storage, _) = transaction!(self, move |tree| {
-            lock::release(tree, txid)?;
+                        tree.insert(value_key, snapshots)?;
+                    }
+                }
 
-            Ok(())
-        })?;
+                Ok(())
+            })
+            .map_err(tx_err_into)
+    }
 
-        try_block!(storage, {
-            if storage.tree.get("gc_lock").map_err(err_into)?.is_some() {
-                return Ok(());
+    pub fn check_retry(
+        &self,
+        tx_result: StdResult<TxPayload, TransactionError<Error>>,
+    ) -> Result<bool> {
+        match tx_result.map_err(tx_err_into) {
+            Err(e) => Err(e),
+            Ok(TxPayload::Success) => Ok(false),
+            Ok(TxPayload::RollbackAndRetry(lock_txid)) => {
+                self.rollback_txid(lock_txid)?;
+
+                self.tree
+                    .transaction(move |tree| lock::release(tree, lock_txid))
+                    .map_err(tx_err_into)
+                    .map(|_| true)
             }
-
-            storage.tree.insert("gc_lock", &[1]).map_err(err_into)?;
-
-            let gc_result = storage.gc();
-
-            storage.tree.remove("gc_lock").map_err(err_into)?;
-
-            gc_result
-        });
-
-        Ok((storage, ()))
+        }
     }
 }
diff --git a/src/translate/function.rs b/src/translate/function.rs
index 13801413..a0665736 100644
--- a/src/translate/function.rs
+++ b/src/translate/function.rs
@@ -78,11 +78,36 @@ pub fn translate_function(sql_function: &SqlFunction) -> Result<Expr> {
 
             Ok(Expr::Function(Box::new(Function::Right { expr, size })))
         }
+        "CEIL" => func_with_one_arg!(Function::Ceil),
+        "ROUND" => func_with_one_arg!(Function::Round),
+        "FLOOR" => func_with_one_arg!(Function::Floor),
         "COUNT" => aggr!(Aggregate::Count),
         "SUM" => aggr!(Aggregate::Sum),
         "MIN" => aggr!(Aggregate::Min),
         "MAX" => aggr!(Aggregate::Max),
         "TRIM" => func_with_one_arg!(Function::Trim),
+        "DIV" => {
+            check_len(name, args.len(), 2)?;
+
+            let dividend = translate_expr(args[0])?;
+            let divisor = translate_expr(args[1])?;
+
+            Ok(Expr::Function(Box::new(Function::Div {
+                dividend,
+                divisor,
+            })))
+        }
+        "MOD" => {
+            check_len(name, args.len(), 2)?;
+
+            let dividend = translate_expr(args[0])?;
+            let divisor = translate_expr(args[1])?;
+
+            Ok(Expr::Function(Box::new(Function::Mod {
+                dividend,
+                divisor,
+            })))
+        }
         _ => Err(TranslateError::UnsupportedFunction(name).into()),
     }
 }


Test patch:
diff --git a/src/tests/function/ceil.rs b/src/tests/function/ceil.rs
new file mode 100644
index 00000000..305dce2d
--- /dev/null
+++ b/src/tests/function/ceil.rs
@@ -0,0 +1,57 @@
+use crate::*;
+
+test_case!(ceil, async move {
+    use Value::{Null, F64};
+
+    let test_cases = vec![
+        (
+            "CREATE TABLE SingleItem (id INTEGER PRIMARY KEY)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"INSERT INTO SingleItem VALUES (0)"#,
+            Ok(Payload::Insert(1)),
+        ),
+        (
+            "SELECT CEIL(0.3) AS ceil1, 
+            CEIL(-0.8) AS ceil2, 
+            CEIL(10) AS ceil3, 
+            CEIL(6.87421) AS ceil4 
+            FROM SingleItem",
+            Ok(select!(
+                "ceil1"        | "ceil2"                   | "ceil3"             | "ceil4";
+                F64            | F64                       | F64                 | F64 ;
+                0.3_f64.ceil()   f64::ceil(-0.8_f64)         f64::from(10).ceil()  6.87421_f64.ceil()
+            )),
+        ),
+        (
+            "SELECT CEIL('string') AS ceil FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("CEIL")).into()),
+        ),
+        (
+            "SELECT CEIL(NULL) AS ceil FROM SingleItem",
+            Ok(select_with_null!(ceil; Null)),
+        ),
+        (
+            "SELECT CEIL(TRUE) AS ceil FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("CEIL")).into()),
+        ),
+        (
+            "SELECT CEIL(FALSE) AS ceil FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("CEIL")).into()),
+        ),
+        (
+            "SELECT CEIL('string', 'string2') AS ceil FROM SingleItem",
+            Err(TranslateError::FunctionArgsLengthNotMatching {
+                name: "CEIL".to_owned(),
+                expected: 1,
+                found: 2,
+            }
+            .into()),
+        ),
+    ];
+
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/function/div_mod.rs b/src/tests/function/div_mod.rs
new file mode 100644
index 00000000..c63a3199
--- /dev/null
+++ b/src/tests/function/div_mod.rs
@@ -0,0 +1,134 @@
+use crate::*;
+
+test_case!(div_mod, async move {
+    use Value::{Null, F64, I64};
+    let eval_div = |dividend, divisor| (dividend / divisor) as i64;
+    let eval_mod = |dividend, divisor| dividend % divisor;
+    let test_cases = vec![
+        (
+            "CREATE TABLE FloatDiv (dividend FLOAT, divisor FLOAT)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"
+            INSERT INTO 
+                FloatDiv (dividend, divisor) 
+            VALUES 
+                (12.0, 3.0), (12.34, 56.78), (-12.3, 4.0)
+            "#,
+            Ok(Payload::Insert(3)),
+        ),
+        (
+            "
+            SELECT 
+                DIV(dividend, divisor),
+                MOD(dividend, divisor) 
+            FROM FloatDiv
+            ",
+            Ok(select!(
+                "DIV(dividend, divisor)" | "MOD(dividend, divisor)"
+                I64                      | F64;
+                eval_div(12.0, 3.0)        eval_mod(12.0, 3.0);
+                eval_div(12.34, 56.78)     eval_mod(12.34, 56.78);
+                eval_div(-12.3, 4.0)       eval_mod(-12.3, 4.0)
+            )),
+        ),
+        (
+            "SELECT DIV(1.0, 0.0) AS quotient FROM FloatDiv",
+            Err(EvaluateError::InvalidDivisorZero.into()),
+        ),
+        (
+            r#"SELECT DIV(1.0, "dividend") AS quotient FROM FloatDiv"#,
+            Err(EvaluateError::FunctionRequiresFloatOrIntegerValue("DIV".to_owned()).into()),
+        ),
+        (
+            "SELECT DIV(1.0) AS quotient FROM FloatDiv",
+            Err(TranslateError::FunctionArgsLengthNotMatching {
+                name: "DIV".to_owned(),
+                expected: 2,
+                found: 1,
+            }
+            .into()),
+        ),
+        (
+            "SELECT MOD(1.0, 2, 3) AS remainder FROM FloatDiv",
+            Err(TranslateError::FunctionArgsLengthNotMatching {
+                name: "MOD".to_owned(),
+                expected: 2,
+                found: 3,
+            }
+            .into()),
+        ),
+        (
+            "CREATE TABLE IntDiv (dividend INTEGER, divisor INTEGER)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"
+            INSERT INTO 
+                IntDiv (dividend, divisor) 
+            VALUES 
+                (12, 3), (12, 7), (12, 34), (-12, 7)
+            "#,
+            Ok(Payload::Insert(4)),
+        ),
+        (
+            "INSERT INTO IntDiv (dividend, divisor) VALUES (12, 2.0)",
+            Err(ValueError::FailedToParseNumber.into()),
+        ),
+        (
+            "
+            SELECT 
+                DIV(dividend, divisor),
+                MOD(dividend, divisor) 
+            FROM IntDiv 
+            ",
+            Ok(select!(
+                "DIV(dividend, divisor)" | "MOD(dividend, divisor)"
+                I64                      | F64;
+                eval_div(12_f64, 3_f64)    eval_mod(12_f64, 3_f64);
+                eval_div(12_f64, 7_f64)    eval_mod(12_f64, 7_f64);
+                eval_div(12_f64, 34_f64)   eval_mod(12_f64, 34_f64);
+                eval_div(-12_f64, 7_f64)   eval_mod(-12_f64, 7_f64)
+            )),
+        ),
+        (
+            "SELECT DIV(1, 0) AS quotient FROM IntDiv",
+            Err(EvaluateError::InvalidDivisorZero.into()),
+        ),
+        (
+            "CREATE TABLE MixDiv (dividend INTEGER NULL, divisor FLOAT NULL)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"
+            INSERT INTO 
+                MixDiv (dividend, divisor) 
+            VALUES 
+                (12, 3.0), (12, 34.0), (12, -5.2),
+                (12, NULL), (NULL, 34.0), (NULL, NULL)
+            "#,
+            Ok(Payload::Insert(6)),
+        ),
+        (
+            "
+            SELECT 
+                DIV(dividend, divisor),
+                MOD(dividend, divisor) 
+            FROM MixDiv 
+            ",
+            Ok(select_with_null!(
+                "DIV(dividend, divisor)"    | "MOD(dividend, divisor)";
+                I64(eval_div(12_f64, 3.0))    F64(eval_mod(12_f64, 3.0));
+                I64(eval_div(12_f64, 34.0))   F64(eval_mod(12_f64, 34.0));
+                I64(eval_div(12_f64, -5.2))   F64(eval_mod(12_f64, -5.2));
+                Null                          Null;
+                Null                          Null;
+                Null                          Null
+            )),
+        ),
+    ];
+    for (sql, expected) in test_cases {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/function/floor.rs b/src/tests/function/floor.rs
new file mode 100644
index 00000000..b6dd2cfc
--- /dev/null
+++ b/src/tests/function/floor.rs
@@ -0,0 +1,59 @@
+use crate::*;
+
+test_case!(floor, async move {
+    use Value::{Null, F64};
+
+    let test_cases = vec![
+        (
+            "CREATE TABLE SingleItem (id INTEGER PRIMARY KEY)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"INSERT INTO SingleItem VALUES (0)"#,
+            Ok(Payload::Insert(1)),
+        ),
+        (
+            r#"
+            SELECT 
+            FLOOR(0.3) as floor1, 
+            FLOOR(-0.8) as floor2, 
+            FLOOR(10) as floor3, 
+            FLOOR(6.87421) as floor4 
+            FROM SingleItem"#,
+            Ok(select!(
+                floor1          | floor2                 | floor3               | floor4
+                F64             | F64                    | F64                  | F64;
+                0.3_f64.floor()   f64::floor(-0.8_f64)     f64::from(10).floor()  6.87421_f64.floor()
+            )),
+        ),
+        (
+            "SELECT FLOOR('string') AS floor FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("FLOOR")).into()),
+        ),
+        (
+            "SELECT FLOOR(NULL) AS floor FROM SingleItem",
+            Ok(select_with_null!(floor; Null)),
+        ),
+        (
+            "SELECT FLOOR(TRUE) AS floor FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("FLOOR")).into()),
+        ),
+        (
+            "SELECT FLOOR(FALSE) AS floor FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("FLOOR")).into()),
+        ),
+        (
+            "SELECT FLOOR('string', 'string2') AS floor FROM SingleItem",
+            Err(TranslateError::FunctionArgsLengthNotMatching {
+                name: "FLOOR".to_owned(),
+                expected: 1,
+                found: 2,
+            }
+            .into()),
+        ),
+    ];
+
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/function/mod.rs b/src/tests/function/mod.rs
index c34f5e47..756131ec 100644
--- a/src/tests/function/mod.rs
+++ b/src/tests/function/mod.rs
@@ -1,4 +1,8 @@
 pub mod cast;
+pub mod ceil;
+pub mod div_mod;
+pub mod floor;
 pub mod left_right;
+pub mod round;
 pub mod trim;
 pub mod upper_lower;
diff --git a/src/tests/function/round.rs b/src/tests/function/round.rs
new file mode 100644
index 00000000..4eb45647
--- /dev/null
+++ b/src/tests/function/round.rs
@@ -0,0 +1,53 @@
+use crate::*;
+
+test_case!(round, async move {
+    use Value::{Null, F64};
+
+    let test_cases = vec![
+        (
+            "CREATE TABLE SingleItem (id INTEGER PRIMARY KEY)",
+            Ok(Payload::Create),
+        ),
+        (
+            r#"INSERT INTO SingleItem VALUES (0)"#,
+            Ok(Payload::Insert(1)),
+        ),
+        (
+            "SELECT ROUND(0.3) AS round1, ROUND(-0.8) AS round2, ROUND(10) AS round3, ROUND(6.87421) AS round4 FROM SingleItem",
+            Ok(select!(
+                round1          | round2                       | round3               | round4
+                F64             | F64                          | F64                  | F64;
+                0.3_f64.round()   f64::round(-0.8_f64)           f64::from(10).round()  6.87421_f64.round()
+            )),
+        ),
+        (
+            "SELECT ROUND('string') AS round FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("ROUND")).into()),
+        ),
+        (
+            "SELECT ROUND(NULL) AS round FROM SingleItem",
+            Ok(select_with_null!(round; Null)),
+        ),
+        (
+            "SELECT ROUND(TRUE) AS round FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("ROUND")).into()),
+        ),
+        (
+            "SELECT ROUND(FALSE) AS round FROM SingleItem",
+            Err(EvaluateError::FunctionRequiresFloatValue(String::from("ROUND")).into()),
+        ),
+        (
+            "SELECT ROUND('string', 'string2') AS round FROM SingleItem",
+            Err(TranslateError::FunctionArgsLengthNotMatching {
+                name: "ROUND".to_owned(),
+                expected: 1,
+                found: 2,
+            }
+            .into()),
+        ),
+    ];
+
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/mod.rs b/src/tests/mod.rs
index 890db14e..1b01f1b9 100644
--- a/src/tests/mod.rs
+++ b/src/tests/mod.rs
@@ -58,8 +58,12 @@ macro_rules! generate_tests {
         glue!(function_upper_lower, function::upper_lower::upper_lower);
         glue!(function_left_right, function::left_right::left_right);
         glue!(function_trim, function::trim::trim);
+        glue!(function_div_mod, function::div_mod::div_mod);
         glue!(function_cast_literal, function::cast::cast_literal);
         glue!(function_cast_value, function::cast::cast_value);
+        glue!(function_ceil, function::ceil::ceil);
+        glue!(function_round, function::round::round);
+        glue!(function_floor, function::floor::floor);
         glue!(join, join::join);
         glue!(join_blend, join::blend);
         glue!(migrate, migrate::migrate);
diff --git a/tests/sled_transaction.rs b/tests/sled_transaction.rs
index 2a8a85d3..fb0e799a 100644
--- a/tests/sled_transaction.rs
+++ b/tests/sled_transaction.rs
@@ -7,8 +7,11 @@
 //! scenarios, but not PHANTOM READ safe.
 
 use {
-    gluesql::{tests::test_indexes, Value::I64, *},
-    std::fs,
+    gluesql::{tests::test_indexes, Value::*, *},
+    std::{
+        fs,
+        time::{SystemTime, UNIX_EPOCH},
+    },
 };
 
 const PATH_PREFIX: &str = "tmp/gluesql";
@@ -57,6 +60,13 @@ fn sled_transaction_basic() {
         glue2 "CREATE TABLE MeTooTheLock (id INTEGER);",
         Err(Error::StorageMsg("database is locked".to_owned()))
     );
+
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "COMMIT;");
+    test!(
+        glue2 "SELECT * FROM AcquireLock;",
+        Err(FetchError::TableNotFound("AcquireLock".to_owned()).into())
+    );
 }
 
 #[test]
@@ -368,13 +378,17 @@ async fn sled_transaction_gc() {
         storage: glue1.storage.map(|mut s| {
             s.state = sled_storage::State::Transaction {
                 txid: 0,
+                created_at: SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_millis(),
                 autocommit: false,
             };
 
             s
         }),
     };
-    test!(glue1 "SELECT * FROM NewGarlic", Err(Error::StorageMsg("fetch failed - expired transaction is used".to_owned())));
+    test!(glue1 "SELECT * FROM NewGarlic", Err(Error::StorageMsg("fetch failed - expired transaction has used (txid)".to_owned())));
     assert_eq!(
         glue1
             .storage
@@ -384,7 +398,328 @@ async fn sled_transaction_gc() {
             .map(|(_, v)| v)
             .map_err(|(_, e)| e),
         Err(Error::StorageMsg(
-            "acquire failed - expired transaction is used".to_owned()
+            "acquire failed - expired transaction has used (txid)".to_owned()
         )),
     );
 }
+
+#[tokio::test]
+async fn sled_transaction_timeout_store() {
+    let path = &format!("{}/transaction_timeout_store", PATH_PREFIX);
+    fs::remove_dir_all(path).unwrap_or(());
+
+    let mut storage1 = SledStorage::new(path).unwrap();
+    storage1.set_transaction_timeout(Some(100));
+    let storage2 = storage1.clone();
+
+    let mut glue1 = Glue::new(storage1);
+    let mut glue2 = Glue::new(storage2);
+
+    let tick = std::time::Duration::from_millis(200);
+    let sleep = || {
+        std::thread::sleep(tick);
+    };
+
+    exec!(glue1 "BEGIN;");
+
+    // glue1 acquires lock
+    exec!(glue1 "CREATE TABLE TxGarlic (id INTEGER);");
+
+    // glue1 lock gets expired due to the timeout
+    sleep();
+
+    // timeout errors
+    test!(glue1 "COMMIT;", Err(Error::StorageMsg("fetch failed - expired transaction has used (timeout)".to_owned())));
+    test!(glue1 "SELECT * FROM TxGarlic", Err(Error::StorageMsg("fetch failed - expired transaction has used (timeout)".to_owned())));
+    assert_eq!(
+        glue1
+            .storage
+            .clone()
+            .unwrap()
+            .update_data("TxGarlic", vec![])
+            .await
+            .map(|(_, v)| v)
+            .map_err(|(_, e)| e),
+        Err(Error::StorageMsg(
+            "acquire failed - expired transaction has used (timeout)".to_owned()
+        )),
+    );
+
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "CREATE TABLE RealGarlic (id INTEGER);");
+    exec!(glue1 "ROLLBACK;");
+    exec!(glue2 "ROLLBACK;");
+
+    // glue2 lock gets expired
+    sleep();
+
+    // glue1 must succeed to create tables: TxGarlic & RealGarlic
+    exec!(glue1 "CREATE TABLE TxGarlic (id2 INTEGER);");
+    exec!(glue1 "CREATE TABLE RealGarlic (id2 INTEGER);");
+
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "INSERT INTO TxGarlic VALUES (10);");
+    sleep();
+    exec!(glue2 "INSERT INTO TxGarlic VALUES (20);");
+    test!(glue1 "COMMIT;", Err(Error::StorageMsg("fetch failed - expired transaction has used (timeout)".to_owned())));
+    exec!(glue1 "ROLLBACK;");
+
+    // glue1 lock has expired, so TxGarlic table has only a single row (20)
+    test!(
+        glue1 "SELECT * FROM TxGarlic;",
+        Ok(select!(id2 I64; 20))
+    );
+
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "UPDATE TxGarlic SET id2 = id2 * 2;");
+    test!(
+        glue1 "SELECT * FROM TxGarlic;",
+        Ok(select!(id2 I64; 40))
+    );
+
+    // glue1 lock gets expired
+    sleep();
+
+    // glue1 tx must rollback
+    test!(
+        glue2 "SELECT * FROM TxGarlic;",
+        Ok(select!(id2 I64; 20))
+    );
+
+    exec!(glue1 "ROLLBACK;");
+    test!(
+        glue1 "SELECT * FROM TxGarlic;",
+        Ok(select!(id2 I64; 20))
+    );
+    test!(
+        glue2 "SELECT * FROM TxGarlic;",
+        Ok(select!(id2 I64; 20))
+    );
+
+    // UPDATE
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "UPDATE TxGarlic SET id2 = id2 + 1;");
+    sleep();
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "UPDATE TxGarlic SET id2 = id2 + 1;");
+    exec!(glue2 "ROLLBACK;");
+    exec!(glue1 "ROLLBACK;");
+
+    // DELETE
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "DELETE FROM TxGarlic;");
+    test!(glue1 "SELECT * FROM TxGarlic;", Ok(select!(id2)));
+    sleep();
+    test!(glue2 "SELECT * FROM TxGarlic;", Ok(select!(id2 I64; 20)));
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "DELETE FROM TxGarlic");
+    exec!(glue1 "ROLLBACK;");
+    exec!(glue2 "ROLLBACK;");
+    test!(glue1 "SELECT * FROM TxGarlic;", Ok(select!(id2 I64; 20)));
+    test!(glue2 "SELECT * FROM TxGarlic;", Ok(select!(id2 I64; 20)));
+
+    // DROP TABLE
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "DROP TABLE TxGarlic;");
+    sleep();
+    test!(glue2 "COMMIT;", Err(Error::StorageMsg("fetch failed - expired transaction has used (timeout)".to_owned())));
+    exec!(glue1 "DROP TABLE TxGarlic;");
+    test!(
+        glue1 "SELECT * FROM TxGarlic;",
+        Err(FetchError::TableNotFound("TxGarlic".to_owned()).into())
+    );
+    exec!(glue2 "ROLLBACK;");
+    test!(
+        glue2 "SELECT * FROM TxGarlic;",
+        Err(FetchError::TableNotFound("TxGarlic".to_owned()).into())
+    );
+}
+
+#[cfg(feature = "alter-table")]
+#[tokio::test]
+async fn sled_transaction_timeout_alter() {
+    let path = &format!("{}/transaction_timeout_alter", PATH_PREFIX);
+    fs::remove_dir_all(path).unwrap_or(());
+
+    let mut storage1 = SledStorage::new(path).unwrap();
+    storage1.set_transaction_timeout(Some(100));
+    let storage2 = storage1.clone();
+
+    let mut glue1 = Glue::new(storage1);
+    let mut glue2 = Glue::new(storage2);
+
+    let tick = std::time::Duration::from_millis(200);
+    let sleep = || {
+        std::thread::sleep(tick);
+    };
+
+    exec!(glue1 "CREATE TABLE TxAlter (id INTEGER, num INTEGER);");
+    exec!(glue1 "INSERT INTO TxAlter VALUES (1, 100);");
+
+    // DROP COLUMN
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "ALTER TABLE TxAlter DROP COLUMN num;");
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(id I64; 1)));
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+    sleep();
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "ALTER TABLE TxAlter DROP COLUMN num;");
+    exec!(glue2 "ROLLBACK;");
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+
+    // ADD COLUMN
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "ALTER TABLE TxAlter ADD COLUMN flag BOOLEAN DEFAULT TRUE;");
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num | flag I64 | I64 | Bool; 1 100 true)));
+
+    exec!(glue1 "ROLLBACK;");
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+    sleep();
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+    exec!(glue2 "ROLLBACK;");
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+
+    // RENAME COLUMN
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "ALTER TABLE TxAlter RENAME COLUMN id TO jd;");
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(jd | num I64 | I64; 1 100)));
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(id | num I64 | I64; 1 100)));
+    sleep();
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "ALTER TABLE TxAlter RENAME COLUMN id TO kd;");
+    exec!(glue2 "COMMIT;");
+    exec!(glue1 "ROLLBACK;");
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(kd | num I64 | I64; 1 100)));
+    test!(glue2 "SELECT * FROM TxAlter;", Ok(select!(kd | num I64 | I64; 1 100)));
+
+    // RENAME TABLE
+    exec!(glue2 "BEGIN;");
+    exec!(glue2 "ALTER TABLE TxAlter RENAME TO TxAltericano;");
+    test!(glue2 "SELECT * FROM TxAltericano;", Ok(select!(kd | num I64 | I64; 1 100)));
+    test!(
+        glue2 "SELECT * FROM TxAlter;",
+        Err(FetchError::TableNotFound("TxAlter".to_owned()).into())
+    );
+    test!(glue1 "SELECT * FROM TxAlter;", Ok(select!(kd | num I64 | I64; 1 100)));
+    test!(
+        glue1 "SELECT * FROM TxAlterericano;",
+        Err(FetchError::TableNotFound("TxAlterericano".to_owned()).into())
+    );
+    sleep();
+    exec!(glue1 "ALTER TABLE TxAlter RENAME TO TxSoprano;");
+    test!(glue1 "SELECT * FROM TxSoprano;", Ok(select!(kd | num I64 | I64; 1 100)));
+    exec!(glue2 "ROLLBACK;");
+    test!(glue2 "SELECT * FROM TxSoprano;", Ok(select!(kd | num I64 | I64; 1 100)));
+}
+
+#[tokio::test]
+async fn sled_transaction_timeout_index() {
+    use ast::IndexOperator::Eq;
+
+    let path = &format!("{}/transaction_timeout_index", PATH_PREFIX);
+    fs::remove_dir_all(path).unwrap_or(());
+
+    let mut storage1 = SledStorage::new(path).unwrap();
+    storage1.set_transaction_timeout(Some(100));
+    let storage2 = storage1.clone();
+
+    let mut glue1 = Glue::new(storage1);
+    let mut glue2 = Glue::new(storage2);
+
+    let tick = std::time::Duration::from_millis(200);
+    let sleep = || {
+        std::thread::sleep(tick);
+    };
+
+    exec!(glue1 "CREATE TABLE TxIndex (id INTEGER);");
+    exec!(glue1 "INSERT INTO TxIndex VALUES (1);");
+
+    // CREATE INDEX
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "CREATE INDEX idx_id ON TxIndex (id);");
+    test_idx!(
+        glue1 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    sleep();
+    exec!(glue2 "CREATE INDEX idx_id ON TxIndex (id);");
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    exec!(glue1 "ROLLBACK;");
+    test_idx!(
+        glue1 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+
+    // DROP INDEX
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "DROP INDEX TxIndex.idx_id;");
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    sleep();
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    exec!(glue1 "ROLLBACK;");
+    test_idx!(
+        glue1 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+
+    // DROP AND DROP INDEX
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "DROP INDEX TxIndex.idx_id;");
+    exec!(glue1 "CREATE INDEX idx_id ON TxIndex (id);");
+    sleep();
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    exec!(glue1 "ROLLBACK;");
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    test_idx!(
+        glue1 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+
+    exec!(glue1 "BEGIN;");
+    exec!(glue1 "DROP INDEX TxIndex.idx_id;");
+    sleep();
+    exec!(glue2 "DROP INDEX TxIndex.idx_id;");
+    sleep();
+    exec!(glue1 "ROLLBACK;");
+    exec!(glue1 "CREATE INDEX idx_id ON TxIndex (id);");
+    test_idx!(
+        glue2 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+    test_idx!(
+        glue1 "SELECT * FROM TxIndex WHERE id = 1;",
+        idx!(idx_id, Eq, "1"),
+        Ok(select!(id I64; 1))
+    );
+}