事务提交时检测写写冲突
This commit is contained in:
LIm1tZz7 committed
1 parent
484b35a8bc
commit
85008895dd
4 files changed
+161
-25
No files matched your search
@@ -28,5 +28,5 @@
|
||||
- 改动说明只在交流中记录。
|
||||
|
||||
## 当前重点
|
||||
- 主线已完成：高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断。
|
||||
- 主线已完成：高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测。
|
||||
- 进度与计数见 `agent_credientials/project-plan.txt`，限制与风险见 `agent_credientials/known-issues.txt`。
|
||||
@@ -264,7 +264,9 @@ commitTransaction session = do
|
||||
liveResult <- beSnapshot backend
|
||||
case liveResult of
|
||||
Left err -> pure (Left (engineErrorCode err))
|
||||
Right live -> do
|
||||
Right live -> case transactionConflict (txBase transaction) (txStaged transaction) live of
|
||||
Just msg -> pure (Left (SessionError "serialization_failure" (T.pack msg)))
|
||||
Nothing -> do
|
||||
applied <- beApplyTransaction backend (transactionOps (txBase transaction) (txStaged transaction) live)
|
||||
case applied of
|
||||
Left err -> pure (Left (engineErrorCode err))
|
||||
@@ -303,45 +305,79 @@ resultOfStaged db statement rows = resultOf (StatementResult cols rows)
|
||||
[] -> columnsFromStatement db statement
|
||||
(r : _) -> map fst r
|
||||
|
||||
-- | 一张表的提交差异与冲突
|
||||
data TableDiff = TableDiff
|
||||
{ tdDeletes :: [Int]
|
||||
, tdWrites :: [Row]
|
||||
, tdReplace :: Maybe [Row]
|
||||
, tdConflict :: Maybe String
|
||||
}
|
||||
|
||||
-- | 提交差分：相对快照列出要删、要写与要替换的行
|
||||
transactionOps :: Database -> Database -> Database -> [TxnOp]
|
||||
transactionOps base staged live =
|
||||
concatMap opsFor names
|
||||
transactionOps base staged live = concatMap opsOf (transactionDiffs base staged live)
|
||||
where
|
||||
-- | 一张表的请求：先删后写
|
||||
opsOf (name, diff) =
|
||||
[TxnDelete name (tdDeletes diff) | not (null (tdDeletes diff))]
|
||||
++ [TxnUpsert name (tdWrites diff) | not (null (tdWrites diff))]
|
||||
++ [TxnReplace name rows | Just rows <- [tdReplace diff]]
|
||||
|
||||
-- | 提交前校验：本事务动过的行是否已被别人改掉
|
||||
transactionConflict :: Database -> Database -> Database -> Maybe String
|
||||
transactionConflict base staged live =
|
||||
case [msg | (_, diff) <- transactionDiffs base staged live, Just msg <- [tdConflict diff]] of
|
||||
(msg : _) -> Just msg
|
||||
[] -> Nothing
|
||||
|
||||
-- | 逐表算提交差异，顺带找出冲突
|
||||
transactionDiffs :: Database -> Database -> Database -> [(String, TableDiff)]
|
||||
transactionDiffs base staged live = [(name, diffFor name) | name <- names]
|
||||
where
|
||||
-- | 事务里碰得着的表：数据语句不增删表
|
||||
names = nubKeys (map fst base ++ map fst staged)
|
||||
|
||||
-- | 一张表的差分
|
||||
opsFor name
|
||||
| all (isJust . rowKey) (baseRows ++ stagedRows) = keyedOps
|
||||
| stagedRows == baseRows = []
|
||||
| otherwise = [TxnReplace name stagedRows]
|
||||
-- | 一张表的差异
|
||||
diffFor name
|
||||
| all (isJust . rowKey) (baseRows ++ stagedRows) = keyedDiff
|
||||
-- 没动过这张表：别人改成什么样都不关本事务
|
||||
| stagedRows == baseRows = emptyDiff
|
||||
-- 动过，库里还是基线：整表替换
|
||||
| liveRows == baseRows = emptyDiff {tdReplace = Just stagedRows}
|
||||
| otherwise = emptyDiff {tdConflict = Just ("write conflict on table " ++ name)}
|
||||
where
|
||||
baseRows = rowsOf name base
|
||||
stagedRows = rowsOf name staged
|
||||
liveRows = rowsOf name live
|
||||
|
||||
baseByKey = [(k, r) | r <- baseRows, Just k <- [rowKey r]]
|
||||
liveByKey = [(k, r) | r <- liveRows, Just k <- [rowKey r]]
|
||||
stagedByKey = [(k, r) | r <- stagedRows, Just k <- [rowKey r]]
|
||||
baseKeys = [k | r <- baseRows, Just k <- [rowKey r]]
|
||||
|
||||
-- 开事务时就存在、现在没了：删；内容变了的行也先删再写
|
||||
gone = [k | (k, _) <- liveByKey, k `elem` baseKeys, not (any ((== k) . fst) stagedByKey)]
|
||||
rewritten = [k | (k, r) <- stagedByKey, Just liveRow <- [lookup k liveByKey], liveRow /= r]
|
||||
deleted = gone ++ rewritten
|
||||
-- 本事务动过的键：内容改过、新增，或者删掉
|
||||
written = [k | (k, r) <- stagedByKey, lookup k baseByKey /= Just r]
|
||||
removed = [k | (k, _) <- baseByKey, not (any ((== k) . fst) stagedByKey)]
|
||||
touched = written ++ removed
|
||||
|
||||
-- 新增的行，以及跟库里不一样的行；并发插进来的行不动
|
||||
written =
|
||||
[ r
|
||||
| (k, r) <- stagedByKey
|
||||
, case lookup k liveByKey of
|
||||
Nothing -> True
|
||||
Just liveRow -> liveRow /= r
|
||||
-- 两边都删了算一致；其余只要跟基线不一样就是冲突
|
||||
victims =
|
||||
[ k
|
||||
| k <- touched
|
||||
, lookup k liveByKey /= lookup k baseByKey
|
||||
, not (k `elem` removed && lookup k liveByKey == Nothing)
|
||||
]
|
||||
|
||||
keyedOps =
|
||||
[TxnDelete name deleted | not (null deleted)]
|
||||
++ [TxnUpsert name written | not (null written)]
|
||||
keyedDiff =
|
||||
emptyDiff
|
||||
{ tdDeletes = [k | k <- touched, isJust (lookup k liveByKey)]
|
||||
, tdWrites = [r | (k, r) <- stagedByKey, k `elem` written]
|
||||
, tdConflict = case victims of
|
||||
(k : _) -> Just ("write conflict on table " ++ name ++ ", id " ++ show k)
|
||||
[] -> Nothing
|
||||
}
|
||||
|
||||
-- | 没有差异的空结果
|
||||
emptyDiff = TableDiff [] [] Nothing Nothing
|
||||
|
||||
-- | 表名去重但保持顺序
|
||||
nubKeys = foldr (\name acc -> if name `elem` acc then acc else name : acc) []
|
||||
|
||||
@@ -652,6 +652,10 @@ sessionAge session wanted = do
|
||||
failureMessage :: Either SessionError QueryResult -> Text
|
||||
failureMessage outcome = either sessMessage (const "") outcome
|
||||
|
||||
-- | 取失败里的错误码
|
||||
failureCode :: Either SessionError QueryResult -> Text
|
||||
failureCode outcome = either sessCode (const "") outcome
|
||||
|
||||
-- | 取结果第一列的整数
|
||||
intValues :: [[Value]] -> [Int]
|
||||
intValues rows = [n | row <- rows, Just (VInt n) <- [nth 0 row]]
|
||||
@@ -713,6 +717,77 @@ transactionSpec = describe "server transactions" $ do
|
||||
denied <- runStatementCoded session "use other"
|
||||
failureMessage denied `shouldBe` "cannot switch database inside a transaction"
|
||||
|
||||
it "refuses COMMIT when another session changed the same row" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "update users set age = 31 where id = 1"
|
||||
_ <- mustSql second "update users set age = 42 where id = 1"
|
||||
_ <- mustSql first "commit"
|
||||
denied <- runStatementCoded second "commit"
|
||||
failureCode denied `shouldBe` "serialization_failure"
|
||||
failureMessage denied `shouldBe` "write conflict on table users, id 1"
|
||||
-- 事务保留：回滚之后看到的还是先提交的那个值
|
||||
_ <- mustSql second "rollback"
|
||||
sessionAge second 1 `shouldReturn` Just 31
|
||||
|
||||
it "refuses COMMIT when the row was deleted meanwhile" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "delete from users where id = 2"
|
||||
_ <- mustSql second "update users set age = 42 where id = 2"
|
||||
_ <- mustSql first "commit"
|
||||
denied <- runStatementCoded second "commit"
|
||||
failureMessage denied `shouldBe` "write conflict on table users, id 2"
|
||||
_ <- mustSql second "rollback"
|
||||
ids <- sessionIds second
|
||||
ids `shouldMatchList` [1, 3, 4, 5]
|
||||
|
||||
it "refuses COMMIT when deleting a row that changed meanwhile" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "update users set age = 31 where id = 3"
|
||||
_ <- mustSql second "delete from users where id = 3"
|
||||
_ <- mustSql first "commit"
|
||||
denied <- runStatementCoded second "commit"
|
||||
failureMessage denied `shouldBe` "write conflict on table users, id 3"
|
||||
_ <- mustSql second "rollback"
|
||||
sessionAge second 3 `shouldReturn` Just 31
|
||||
|
||||
it "refuses COMMIT when another session inserted the same id" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "insert into users (id, name, age) values (7, 'seven', 27)"
|
||||
_ <- mustSql second "insert into users (id, name, age) values (7, 'other', 70)"
|
||||
_ <- mustSql first "commit"
|
||||
denied <- runStatementCoded second "commit"
|
||||
failureMessage denied `shouldBe` "write conflict on table users, id 7"
|
||||
_ <- mustSql second "rollback"
|
||||
ids <- sessionIds second
|
||||
ids `shouldMatchList` [1, 2, 3, 4, 5, 7]
|
||||
|
||||
it "keeps other sessions' rows untouched when the write sets do not overlap" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "insert into users (id, name, age) values (6, 'six', 26)"
|
||||
_ <- mustSql first "update users set age = 31 where id = 1"
|
||||
_ <- mustSql second "update users set age = 42 where id = 2"
|
||||
_ <- mustSql first "commit"
|
||||
_ <- mustSql second "commit"
|
||||
sessionAge second 1 `shouldReturn` Just 31
|
||||
sessionAge second 2 `shouldReturn` Just 42
|
||||
ids <- sessionIds second
|
||||
ids `shouldMatchList` [1, 2, 3, 4, 5, 6]
|
||||
|
||||
it "lets both sessions delete the same row" $ withTwoSessions $ \first second -> do
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "delete from users where id = 4"
|
||||
_ <- mustSql second "delete from users where id = 4"
|
||||
_ <- mustSql first "commit"
|
||||
_ <- mustSql second "commit"
|
||||
ids <- sessionIds second
|
||||
ids `shouldMatchList` [1, 2, 3, 5]
|
||||
|
||||
it "needs a current database to start" $ do
|
||||
db <- newMVar testDb
|
||||
settings <- tempSettingsPath "transaction-bare"
|
||||
@@ -807,3 +882,28 @@ ipcConcurrencySpec = describe "server backend on the real storage" $ do
|
||||
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")
|
||||
let alpha = beWithDatabase base "alpha"
|
||||
mustRun (beStatement alpha "CREATE TABLE users (id int, name text, age int)")
|
||||
mustRun (beStatement alpha "INSERT INTO users (id, name, age) VALUES (1, 'a', 21)")
|
||||
settings <- tempSettingsPath "transaction-ipc-conflict"
|
||||
removeIfExists settings
|
||||
first <- newSession alpha (T.pack testDatabaseName) settings
|
||||
second <- newSession alpha (T.pack testDatabaseName) settings
|
||||
_ <- mustSql first "use alpha"
|
||||
_ <- mustSql second "use alpha"
|
||||
_ <- mustSql first "begin"
|
||||
_ <- mustSql second "begin"
|
||||
_ <- mustSql first "update users set age = 31 where id = 1"
|
||||
_ <- mustSql second "update users set age = 42 where id = 1"
|
||||
_ <- mustSql first "commit"
|
||||
denied <- runStatementCoded second "commit"
|
||||
failureCode denied `shouldBe` "serialization_failure"
|
||||
failureMessage denied `shouldBe` "write conflict on table users, id 1"
|
||||
_ <- mustSql second "rollback"
|
||||
rows <- rowsOf (beStatement alpha "SELECT id, age FROM users")
|
||||
intColumn "id" rows `shouldBe` [1]
|
||||
intColumn "age" rows `shouldBe` [31]
|
||||
|
||||
@@ -49,7 +49,7 @@ chusql-server
|
||||
|
||||
语句里的库前缀和表限定符由引擎在语义层解析：存储层只接受精确的表名键，因此 `FROM sales.orders` 与选定 `sales` 后的 `FROM orders` 指向同一张表。
|
||||
|
||||
显式事务在会话层实现：`BEGIN` 把当前数据库整库读成内存快照，事务内的语句都在快照上执行；`COMMIT` 计算快照与库的差分，用一条 `apply_transaction` 请求交给存储层应用；`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交：事务里看不到别的事务未提交或后来提交的改动，提交对外一次性可见。
|
||||
显式事务在会话层实现：`BEGIN` 把当前数据库整库读成内存快照，事务内的语句都在快照上执行；`COMMIT` 先把快照、基线与库里现值按行 `id` 逐键比对，检测到写写冲突就报 `serialization_failure` 并保留事务；没有冲突才把差分用一条 `apply_transaction` 请求交给存储层应用；`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交：事务里看不到别的事务未提交或后来提交的改动，提交对外一次性可见，冲突由调用方回滚或重试。
|
||||
|
||||
WAL 负责崩溃恢复：每条记录带自增 LSN，一组写操作与一条提交标记一次落盘；每组提交在数据和字典都刷盘之后把 `wal.checkpoint` 推进到这条提交的 LSN，含义是「到此为止的改动都已落盘」，再按这个边界丢掉日志前缀——所以日志里通常只有当组在飞的记录。进程重启时只重放检查点之后的记录，按标记分组取已提交的组、丢弃没有标记的整组，重放前逐条做幂等检查作为第二道保险，结束后再把检查点推到最后一条提交。LSN 从检查点与文件尾里较大的那个续号，跨重启单调，`SystemStatus` 应答里的 `last_lsn` 报告该检查点 LSN。建表、删表、建索引、删索引与数据改动走同一条 WAL，恢复时一并重放。
|
||||
|
||||
|
||||
Reference in new issue
Block a user