diff --git a/AGENTS.md b/AGENTS.md index e6d2c1a..bcdb0c5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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`。 diff --git a/chusql-server/src/ChuSQL/Server/Session.hs b/chusql-server/src/ChuSQL/Server/Session.hs index 8d13334..99ce9a3 100644 --- a/chusql-server/src/ChuSQL/Server/Session.hs +++ b/chusql-server/src/ChuSQL/Server/Session.hs @@ -264,13 +264,15 @@ commitTransaction session = do liveResult <- beSnapshot backend case liveResult of Left err -> pure (Left (engineErrorCode err)) - Right live -> do - applied <- beApplyTransaction backend (transactionOps (txBase transaction) (txStaged transaction) live) - case applied of - Left err -> pure (Left (engineErrorCode err)) - Right () -> do - writeIORef (ssTransaction session) Nothing - pure (Right (emptyResult (Just current))) + 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)) + Right () -> do + writeIORef (ssTransaction session) Nothing + pure (Right (emptyResult (Just current))) -- | 回滚:快照丢掉,存储层没被改过 rollbackTransaction :: Session -> IO (Either SessionError QueryResult) @@ -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) [] diff --git a/chusql-server/test/Spec.hs b/chusql-server/test/Spec.hs index ddc80e0..b3c21a6 100644 --- a/chusql-server/test/Spec.hs +++ b/chusql-server/test/Spec.hs @@ -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] + diff --git a/docs/architecture.md b/docs/architecture.md index e96ef7e..4bbdec2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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,恢复时一并重放。