事务保存点
This commit is contained in:
1 parent
85008895dd
commit
e2997627a7
10 files changed
+171
-6
No files matched your search
@@ -69,10 +69,11 @@ data Session = Session
|
||||
, ssTransaction :: IORef (Maybe Transaction)
|
||||
}
|
||||
|
||||
-- | 一条会话里的显式事务:开事务时的快照与事务内改到的状态
|
||||
-- | 一条会话里的显式事务:开事务时的快照、事务内改到的状态与保存点栈
|
||||
data Transaction = Transaction
|
||||
{ txBase :: Database
|
||||
, txStaged :: Database
|
||||
, txSavepoints :: [(String, Database)]
|
||||
}
|
||||
|
||||
-- | 开一个会话:账号与角色服务、配置策略都和 Web 端一致
|
||||
@@ -191,6 +192,9 @@ runStatementCoded session sql = case parseStatement (T.unpack sql) of
|
||||
Right BeginTransaction -> beginTransaction session
|
||||
Right CommitTransaction -> commitTransaction session
|
||||
Right RollbackTransaction -> rollbackTransaction session
|
||||
Right (Savepoint name) -> savepointTransaction session (T.pack name)
|
||||
Right (RollbackToSavepoint name) -> rollbackToSavepoint session (T.pack name)
|
||||
Right (ReleaseSavepoint name) -> releaseSavepoint session (T.pack name)
|
||||
Right statement
|
||||
| Just command <- accountCommand statement -> runAccount session command
|
||||
| Just command <- privilegeCommand statement -> runPrivilege session command
|
||||
@@ -249,7 +253,7 @@ beginTransaction session = do
|
||||
case snapshotResult of
|
||||
Left err -> pure (Left (engineErrorCode err))
|
||||
Right db -> do
|
||||
writeIORef (ssTransaction session) (Just (Transaction db db))
|
||||
writeIORef (ssTransaction session) (Just (Transaction db db []))
|
||||
pure (Right (emptyResult (Just current)))
|
||||
|
||||
-- | 提交:把快照差分交给存储层应用,成功才结束事务
|
||||
@@ -285,6 +289,54 @@ rollbackTransaction session = do
|
||||
writeIORef (ssTransaction session) Nothing
|
||||
pure (Right (emptyResult (Just current)))
|
||||
|
||||
-- | 建保存点:把事务当前状态压进保存点栈
|
||||
savepointTransaction :: Session -> Text -> IO (Either SessionError QueryResult)
|
||||
savepointTransaction session name = do
|
||||
current <- readIORef (ssCurrent session)
|
||||
existing <- readIORef (ssTransaction session)
|
||||
case existing of
|
||||
Nothing -> pure (Left (SessionError "query_error" "no transaction in progress"))
|
||||
Just transaction -> do
|
||||
let stack = txSavepoints transaction ++ [(T.unpack name, txStaged transaction)]
|
||||
writeIORef (ssTransaction session) (Just transaction {txSavepoints = stack})
|
||||
pure (Right (emptyResult (Just current)))
|
||||
|
||||
-- | 回滚到保存点:事务状态退回保存点,该保存点保留
|
||||
rollbackToSavepoint :: Session -> Text -> IO (Either SessionError QueryResult)
|
||||
rollbackToSavepoint session name = do
|
||||
current <- readIORef (ssCurrent session)
|
||||
existing <- readIORef (ssTransaction session)
|
||||
case existing of
|
||||
Nothing -> pure (Left (SessionError "query_error" "no transaction in progress"))
|
||||
Just transaction -> case savepointIndex name (txSavepoints transaction) of
|
||||
Nothing -> pure (Left (SessionError "query_error" ("no such savepoint: " <> name)))
|
||||
Just index -> do
|
||||
let (_, staged) = txSavepoints transaction !! index
|
||||
writeIORef
|
||||
(ssTransaction session)
|
||||
(Just transaction {txStaged = staged, txSavepoints = take (index + 1) (txSavepoints transaction)})
|
||||
pure (Right (emptyResult (Just current)))
|
||||
|
||||
-- | 释放保存点:该保存点与它之后的保存点一起丢掉
|
||||
releaseSavepoint :: Session -> Text -> IO (Either SessionError QueryResult)
|
||||
releaseSavepoint session name = do
|
||||
current <- readIORef (ssCurrent session)
|
||||
existing <- readIORef (ssTransaction session)
|
||||
case existing of
|
||||
Nothing -> pure (Left (SessionError "query_error" "no transaction in progress"))
|
||||
Just transaction -> case savepointIndex name (txSavepoints transaction) of
|
||||
Nothing -> pure (Left (SessionError "query_error" ("no such savepoint: " <> name)))
|
||||
Just index -> do
|
||||
writeIORef (ssTransaction session) (Just transaction {txSavepoints = take index (txSavepoints transaction)})
|
||||
pure (Right (emptyResult (Just current)))
|
||||
|
||||
-- | 最近一次同名保存点的下标
|
||||
savepointIndex :: Text -> [(String, Database)] -> Maybe Int
|
||||
savepointIndex name entries =
|
||||
case [index | (index, (label, _)) <- zip [0 ..] entries, T.pack label == name] of
|
||||
[] -> Nothing
|
||||
hits -> Just (last hits)
|
||||
|
||||
-- | 事务里跑一条数据语句:只动内存快照,不碰存储层
|
||||
runInTransaction :: Session -> Statement -> IO (Either SessionError QueryResult)
|
||||
runInTransaction session statement = do
|
||||
|
||||
@@ -788,6 +788,66 @@ transactionSpec = describe "server transactions" $ do
|
||||
ids <- sessionIds second
|
||||
ids `shouldMatchList` [1, 2, 3, 5]
|
||||
|
||||
it "keeps the work before a savepoint and drops the work after it" $ withMemorySession $ \session -> do
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "update users set age = 31 where id = 1"
|
||||
_ <- mustSql session "savepoint spot"
|
||||
_ <- mustSql session "insert into users (id, name, age) values (6, 'six', 26)"
|
||||
_ <- mustSql session "delete from users where id = 2"
|
||||
_ <- mustSql session "rollback to spot"
|
||||
inside <- sessionIds session
|
||||
inside `shouldMatchList` [1, 2, 3, 4, 5]
|
||||
sessionAge session 1 `shouldReturn` Just 31
|
||||
_ <- mustSql session "commit"
|
||||
ids <- sessionIds session
|
||||
ids `shouldMatchList` [1, 2, 3, 4, 5]
|
||||
sessionAge session 1 `shouldReturn` Just 31
|
||||
|
||||
it "lets a savepoint be used again after rolling back to it" $ withMemorySession $ \session -> do
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "insert into users (id, name, age) values (6, 'six', 26)"
|
||||
_ <- mustSql session "savepoint spot"
|
||||
_ <- mustSql session "insert into users (id, name, age) values (7, 'seven', 27)"
|
||||
_ <- mustSql session "rollback to spot"
|
||||
_ <- mustSql session "insert into users (id, name, age) values (8, 'eight', 28)"
|
||||
ids <- sessionIds session
|
||||
ids `shouldMatchList` [1, 2, 3, 4, 5, 6, 8]
|
||||
|
||||
it "drops the savepoints above the one rolled back to" $ withMemorySession $ \session -> do
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "savepoint one"
|
||||
_ <- mustSql session "savepoint two"
|
||||
_ <- mustSql session "rollback to one"
|
||||
missing <- runStatementCoded session "rollback to two"
|
||||
failureMessage missing `shouldBe` "no such savepoint: two"
|
||||
|
||||
it "drops a released savepoint and the ones after it" $ withMemorySession $ \session -> do
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "savepoint one"
|
||||
_ <- mustSql session "savepoint two"
|
||||
released <- mustSql session "release one"
|
||||
qrRowCount released `shouldBe` 0
|
||||
first <- runStatementCoded session "rollback to one"
|
||||
failureMessage first `shouldBe` "no such savepoint: one"
|
||||
second <- runStatementCoded session "rollback to two"
|
||||
failureMessage second `shouldBe` "no such savepoint: two"
|
||||
|
||||
it "clears the savepoints on COMMIT" $ withMemorySession $ \session -> do
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "savepoint spot"
|
||||
_ <- mustSql session "commit"
|
||||
missing <- runStatementCoded session "rollback to spot"
|
||||
failureMessage missing `shouldBe` "no transaction in progress"
|
||||
|
||||
it "needs a transaction and a known name for savepoints" $ withMemorySession $ \session -> do
|
||||
outside <- runStatementCoded session "savepoint spot"
|
||||
failureMessage outside `shouldBe` "no transaction in progress"
|
||||
_ <- mustSql session "begin"
|
||||
unknown <- runStatementCoded session "rollback to nowhere"
|
||||
failureMessage unknown `shouldBe` "no such savepoint: nowhere"
|
||||
gone <- runStatementCoded session "release nowhere"
|
||||
failureMessage gone `shouldBe` "no such savepoint: nowhere"
|
||||
|
||||
it "needs a current database to start" $ do
|
||||
db <- newMVar testDb
|
||||
settings <- tempSettingsPath "transaction-bare"
|
||||
@@ -882,6 +942,26 @@ ipcConcurrencySpec = describe "server backend on the real storage" $ do
|
||||
rows <- rowsOf (beStatement alpha "SELECT id FROM t")
|
||||
intColumn "id" rows `shouldBe` [1]
|
||||
|
||||
it "rolls back to a savepoint on the real storage" $ withIpcStorage $ do
|
||||
base <- ipcBackend
|
||||
mustRun (beStatement base "CREATE DATABASE alpha")
|
||||
let alpha = beWithDatabase base "alpha"
|
||||
settings <- tempSettingsPath "transaction-ipc-savepoint"
|
||||
removeIfExists settings
|
||||
session <- newSession alpha (T.pack testDatabaseName) settings
|
||||
_ <- mustSql session "use alpha"
|
||||
_ <- mustSql session "create table t (id int)"
|
||||
_ <- mustSql session "begin"
|
||||
_ <- mustSql session "insert into t (id) values (1)"
|
||||
_ <- mustSql session "savepoint spot"
|
||||
_ <- mustSql session "insert into t (id) values (2)"
|
||||
_ <- mustSql session "rollback to spot"
|
||||
inside <- mustSql session "select * from t"
|
||||
intValues (qrRows inside) `shouldBe` [1]
|
||||
_ <- mustSql session "commit"
|
||||
rows <- rowsOf (beStatement alpha "SELECT id FROM t")
|
||||
intColumn "id" rows `shouldBe` [1]
|
||||
|
||||
it "refuses the second COMMIT on the real storage" $ withIpcStorage $ do
|
||||
base <- ipcBackend
|
||||
mustRun (beStatement base "CREATE DATABASE alpha")
|
||||
|
||||
Reference in new issue
Block a user