From fc9bbc20e4729838ae72826cfeff8708d6b768c5 Mon Sep 17 00:00:00 2001 From: LIm1tZz7 <36764786+TheBadRoger@users.noreply.github.com> Date: Fri, 9 Oct 2026 09:27:44 +0800 Subject: [PATCH] =?UTF-8?q?v26.10.2=20=E4=BF=AE=E5=A4=8Dbug=EF=BC=8C?= =?UTF-8?q?=E7=AE=80=E5=8C=96=E9=A1=B9=E7=9B=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 2 + chusql-core/engine/src/ChuSQL/Core/Engine.hs | 6 +- .../engine/src/ChuSQL/Core/Engine/Parallel.hs | 46 ++-- .../ChuSQL/Core/Engine/Runtime/Functions.hs | 7 +- .../engine/src/ChuSQL/Core/Engine/Storage.hs | 6 +- .../src/ChuSQL/Core/Engine/Storage/FFI.hs | 51 ++--- .../src/ChuSQL/Core/Engine/Storage/IPC.hs | 105 +++++---- .../src/ChuSQL/Core/Engine/Storage/Memory.hs | 4 +- chusql-core/engine/test/NormalizationSpec.hs | 26 +-- chusql-core/engine/test/RuntimeSpec.hs | 1 - chusql-core/engine/test/Spec.hs | 209 ++++++++++++++---- chusql-core/storage/src/btree.rs | 68 +++--- chusql-core/storage/src/config.rs | 45 +--- chusql-core/storage/src/runtime.rs | 23 +- chusql-core/storage/tests/btree_test.rs | 59 +++++ chusql-core/storage/tests/config_test.rs | 7 +- chusql-core/storage/tests/runtime_test.rs | 32 +++ chusql-server/src/ChuSQL/Server/Backend.hs | 10 +- chusql-server/src/ChuSQL/Server/TCP.hs | 44 +++- chusql-server/test/Spec.hs | 36 ++- chusql-web/static/api.js | 22 +- chusql-web/test/runtime-values.mjs | 38 +++- docs/architecture.md | 2 +- docs/normalization.md | 2 +- docs/release-signing.md | 2 +- gate-check/README.md | 3 +- 26 files changed, 557 insertions(+), 299 deletions(-) diff --git a/.gitignore b/.gitignore index a6b55e9..ddf92ec 100644 --- a/.gitignore +++ b/.gitignore @@ -78,7 +78,9 @@ resources/chusql.ui.settings.json .\#* AGENTS.md agent-credientials +skills-lock.json .dsh +.agents # 注意：.zed/tasks.json 是本仓库共享的构建任务（stack build / run / test / repl）， # 需要入库，所以这里不忽略 .zed/ diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine.hs b/chusql-core/engine/src/ChuSQL/Core/Engine.hs index 3890494..5db0c3f 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine.hs @@ -32,10 +32,10 @@ runStatementM (DropDatabase name) = fmap (fmap (const [])) (dropDatabase name) runStatementM (UseDatabase name) = fmap (fmap (const [])) (useDatabase name) runStatementM ShowDatabases = fmap (fmap (map (\name -> [("database", VStr name)]))) listDatabases runStatementM q = do - db <- schema - case prepare db q of + result <- schema + case result >>= \db -> (,) db <$> prepare db q of Left err -> pure (Left err) - Right resolved -> runStatementUncheckedM db resolved + Right (db, resolved) -> runStatementUncheckedM db resolved -- | 按语句类型分发（已检查过） runStatementUncheckedM :: (MonadStorage m) => Database -> Statement -> m (Either String [Row]) diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Parallel.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Parallel.hs index 402369a..71d7227 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Parallel.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Parallel.hs @@ -6,8 +6,10 @@ module ChuSQL.Core.Engine.Parallel ( ) where import Control.Concurrent (forkIO) -import Control.Concurrent.MVar (MVar, modifyMVar, modifyMVar_, newEmptyMVar, newMVar, putMVar, takeMVar) +import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) +import Control.Concurrent.MVar (MVar, modifyMVar, newEmptyMVar, newMVar, putMVar, takeMVar) import Control.Exception (SomeException, throwIO, try) +import Control.Monad (forever) import System.IO.Unsafe (unsafePerformIO) -- 并行执行：常驻工作线程池，结果按提交顺序收回。 @@ -16,11 +18,8 @@ import System.IO.Unsafe (unsafePerformIO) parallelShardLimit :: Int parallelShardLimit = 8 --- | 池：待办队列与唤醒信号 -data Pool = Pool - { poolJobs :: MVar [IO ()] - , poolReady :: MVar () - } +-- | 池：工作线程共享的待办队列 +newtype Pool = Pool (Chan (IO ())) {-# NOINLINE poolRef #-} -- | 进程全局的池，按容量复用 @@ -31,11 +30,14 @@ poolRef = unsafePerformIO (newMVar Nothing) workerPool :: Int -> IO Pool workerPool size = modifyMVar poolRef reuse where - -- | 现有池够大就留着，否则换新的 + -- | 复用队列并按需补充工作线程 reuse current = case current of - Just (capacity, pool) + Just (capacity, pool@(Pool jobs)) | capacity >= wanted -> pure (current, pool) - _ -> do + | otherwise -> do + sequence_ [forkIO (worker jobs) | _ <- [capacity + 1 .. wanted]] + pure (Just (wanted, pool), pool) + Nothing -> do pool <- startPool wanted pure (Just (wanted, pool), pool) @@ -44,29 +46,17 @@ workerPool size = modifyMVar poolRef reuse -- | 起 wanted 个常驻线程吃队列 startPool n = do - jobs <- newMVar [] - ready <- newMVar () - sequence_ [forkIO (worker jobs ready) | _ <- [1 .. n]] - pure (Pool jobs ready) + jobs <- newChan + sequence_ [forkIO (worker jobs) | _ <- [1 .. n]] + pure (Pool jobs) --- | 一个工作线程：被唤醒一次就取一个待办 -worker :: MVar [IO ()] -> MVar () -> IO () -worker jobs ready = do - takeMVar ready - job <- modifyMVar jobs pick - case job of - Nothing -> worker jobs ready - Just act -> act >> worker jobs ready - where - -- | 取队首，空队列就回 Nothing - pick [] = pure ([], Nothing) - pick (j : js) = pure (js, Just j) +-- | 一个工作线程：等待并执行队首待办 +worker :: Chan (IO ()) -> IO () +worker jobs = forever (readChan jobs >>= id) -- | 往池里丢一个待办 submit :: Pool -> IO () -> IO () -submit pool job = do - modifyMVar_ (poolJobs pool) (pure . (++ [job])) - putMVar (poolReady pool) () +submit (Pool jobs) = writeChan jobs -- | 跑一个动作，结果或异常都填进格子 fillCell :: MVar (Either SomeException a) -> IO a -> IO () diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Runtime/Functions.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Runtime/Functions.hs index 21eed20..28242b7 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Runtime/Functions.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Runtime/Functions.hs @@ -32,12 +32,7 @@ builtinFunctions types = do -- | 按名称和完整参数类型解析函数 resolveFunction :: FunctionRegistry -> String -> [TypeId] -> Either String FunctionSignature -resolveFunction (FunctionRegistry functions) name arguments = - case [sig | (sig, _) <- functions, map toLower (signatureName sig) == map toLower name - , signatureArguments sig == arguments] of - [sig] -> Right sig - [] -> Left ("no matching function signature: " ++ name) - _ -> Left ("ambiguous function signature: " ++ name) +resolveFunction registry name arguments = resolveNullableFunction registry name (map Just arguments) -- | 按可空 SQL 参数解析确定签名 resolveNullableFunction :: FunctionRegistry -> String -> [Maybe TypeId] -> Either String FunctionSignature diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage.hs index a2291ce..55154f4 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage.hs @@ -108,11 +108,11 @@ class (Monad m) => MonadStorage m where replaceSchema :: String -> [(String, Column)] -> [Row] -> m (Either String ()) -- \| 取全库快照（表名 + 列 + 全部行） - snapshot :: m Database + snapshot :: m (Either String Database) -- \| 取全库结构：只要表名和列，不要行 - schema :: m Database - schema = fmap (map withoutRows) snapshot + schema :: m (Either String Database) + schema = fmap (fmap (map withoutRows)) snapshot where -- | 表定义去掉所有行 withoutRows (name, t) = (name, t {tableRows = []}) diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/FFI.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/FFI.hs index 3b51a44..cb119dd 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/FFI.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/FFI.hs @@ -10,7 +10,8 @@ module ChuSQL.Core.Engine.Storage.FFI ( maintainStorage, ) where -import Control.Exception (SomeException, try) +import Control.Concurrent (rtsSupportsBoundThreads, runInBoundThread) +import Control.Exception (SomeException, finally, mask_, try) import Data.ByteString qualified as BS import Data.Text qualified as T import Data.Text.Encoding qualified as TE @@ -19,6 +20,7 @@ import Foreign import Foreign.C -- Rust 存储库 cdylib 的 C ABI 绑定（ffi.rs），收发单行 JSON。 +-- 错误读取绑定同一 OS 线程；响应复制期间屏蔽取消并归还缓冲。 -- | 存储实例句柄（Rust 侧的 *mut Storage） newtype StorageHandle = StorageHandle (Ptr ()) @@ -46,15 +48,8 @@ foreign import ccall safe "chusql_storage_maintenance" -- | 离线修复系统目录并返回结果 maintainStorage :: FilePath -> BS.ByteString -> IO (Either String BS.ByteString) -maintainStorage path payload = withCString path $ \config -> - BS.useAsCStringLen payload $ \(input, size) -> alloca $ \outRef -> alloca $ \lenRef -> do - code <- cStorageMaintenance config input (fromIntegral size) outRef lenRef - if code /= 0 then Left <$> lastError else do - output <- peek outRef - len <- peek lenRef - bytes <- BS.packCStringLen (output, fromIntegral len) - cStorageFree output len - pure (Right bytes) +maintainStorage path payload = BS.useAsCString (TE.encodeUtf8 (T.pack path)) $ \config -> + callStorage (cStorageMaintenance config) payload -- | 库版本（Rust 侧的 CARGO_PKG_VERSION） storageVersion :: IO String @@ -62,7 +57,7 @@ storageVersion = cStorageVersion >>= peekCString -- | 打开存储实例，失败给错误文本 openStorage :: Maybe FilePath -> IO (Either String StorageHandle) -openStorage path = do +openStorage path = onStorageThread $ mask_ $ do result <- try (withCStringOrNull path cStorageOpen) :: IO (Either SomeException (Ptr ())) case result of Left e -> pure (Left ("storage library call failed: " ++ show e)) @@ -72,7 +67,7 @@ openStorage path = do where -- | 没有路径就传空指针，否则转成 C 字符串 withCStringOrNull Nothing f = f nullPtr - withCStringOrNull (Just s) f = withCString s f + withCStringOrNull (Just s) f = BS.useAsCString (TE.encodeUtf8 (T.pack s)) f -- | 关掉存储实例 closeStorage :: StorageHandle -> IO () @@ -80,19 +75,25 @@ closeStorage (StorageHandle handle) = cStorageClose handle -- | 一条 JSON 请求换一条 JSON 响应 storageRequest :: StorageHandle -> BS.ByteString -> IO (Either String BS.ByteString) -storageRequest (StorageHandle handle) payload = - BS.useAsCStringLen payload $ \(inPtr, inLen) -> - alloca $ \outPtrRef -> - alloca $ \outLenRef -> do - code <- cStorageRequest handle (castPtr inPtr) (fromIntegral inLen) outPtrRef outLenRef - if code /= 0 - then Left <$> lastError - else do - outPtr <- peek outPtrRef - outLen <- peek outLenRef - bytes <- BS.packCStringLen (castPtr outPtr, fromIntegral outLen) - cStorageFree outPtr outLen - pure (Right bytes) +storageRequest (StorageHandle handle) = callStorage (cStorageRequest handle) + +-- | 调用存储并复制和归还响应缓冲 +callStorage :: (Ptr CChar -> CSize -> Ptr (Ptr CChar) -> Ptr CSize -> IO CInt) + -> BS.ByteString -> IO (Either String BS.ByteString) +callStorage call payload = onStorageThread $ mask_ $ + BS.useAsCStringLen payload $ \(input, size) -> + alloca $ \outRef -> alloca $ \lenRef -> do + code <- call input (fromIntegral size) outRef lenRef + if code /= 0 then Left <$> lastError else do + output <- peek outRef + len <- peek lenRef + bytes <- BS.packCStringLen (output, fromIntegral len) + `finally` cStorageFree output len + pure (Right bytes) + +-- | 多线程运行时绑定存储调用线程 +onStorageThread :: IO a -> IO a +onStorageThread = if rtsSupportsBoundThreads then runInBoundThread else id -- | 当前线程最近一次失败的文本 lastError :: IO String diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs index b7c46b4..28e0f6a 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs @@ -36,7 +36,8 @@ import ChuSQL.Core.Engine.Parallel (parallelShardLimit, poolRun, workerPool) import ChuSQL.Core.Engine.Storage import ChuSQL.Core.Engine.Storage.FFI (closeStorage, openStorage, storageRequest) import Control.Concurrent (getNumCapabilities) -import Control.Concurrent.MVar (MVar, modifyMVar_, newMVar, readMVar) +import Control.Concurrent.MVar (MVar, modifyMVar_, newMVar, putMVar, readMVar, takeMVar, withMVar) +import Control.Exception (bracket, mask_) import Data.Aeson (eitherDecodeStrict, encode) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BL @@ -89,9 +90,33 @@ data StorageLink = StorageLink linkRef :: MVar (Maybe StorageLink) linkRef = unsafePerformIO (newMVar Nothing) +{-# NOINLINE linkUsers #-} +-- | 在途请求计数与空闲信号 +linkUsers :: (MVar Int, MVar ()) +linkUsers = unsafePerformIO $ (,) <$> newMVar 0 <*> newMVar () + +-- | 登记请求并在结束时释放链路占用 +withStorageLink :: (Maybe StorageLink -> IO a) -> IO a +withStorageLink = bracket enter leave + where + (users, idle) = linkUsers + -- | 登记当前链路的在途请求 + enter = withMVar linkRef $ \current -> do + case current of + Nothing -> pure () + Just _ -> modifyMVar_ users $ \count -> do + if count == 0 then takeMVar idle else pure () + pure (count + 1) + pure current + -- | 最后一个请求结束后发出空闲信号 + leave Nothing = pure () + leave (Just _) = mask_ $ modifyMVar_ users $ \count -> do + if count == 1 then putMVar idle () else pure () + pure (count - 1) + -- | 装入一条链路；一个进程只装一次 setStorageLink :: StorageLink -> IO () -setStorageLink link = modifyMVar_ linkRef (const (pure (Just link))) +setStorageLink link = modifyMVar_ linkRef $ \_ -> withMVar (snd linkUsers) (const (pure (Just link))) -- | 打开本地的存储动态库，得到一条链路 localStorageLink :: Maybe FilePath -> IO (Either String StorageLink) @@ -106,7 +131,7 @@ localStorageLink path = fmap (fmap toLink) (openStorage path) -- | 关掉当前链路（进程退出前调用）；没有链路就什么都不做 closeConnection :: IO () -closeConnection = modifyMVar_ linkRef $ \current -> do +closeConnection = modifyMVar_ linkRef $ \current -> withMVar (snd linkUsers) $ \_ -> do case current of Nothing -> pure () Just link -> slClose link @@ -114,22 +139,19 @@ closeConnection = modifyMVar_ linkRef $ \current -> do -- | 发一条请求并解码应答；链路没装就报错，不崩 sendRequest :: Request -> IO Response -sendRequest req = do - current <- readMVar linkRef - case current of - Nothing -> pure (RespError "storage not configured") - Just link -> do - raw <- slRequest link (BL.toStrict (encode req)) - case raw of - Left e -> pure (RespError ("storage error: " ++ e)) - Right bytes -> case eitherDecodeStrict (BS.dropWhileEnd (== 13) bytes) of - Left e -> pure (RespError ("decode error: " ++ e)) - Right resp -> pure resp +sendRequest req = withStorageLink $ \current -> case current of + Nothing -> pure (RespError "storage not configured") + Just link -> do + raw <- slRequest link (BL.toStrict (encode req)) + case raw of + Left e -> pure (RespError ("storage error: " ++ e)) + Right bytes -> case eitherDecodeStrict (BS.dropWhileEnd (== 13) bytes) of + Left e -> pure (RespError ("decode error: " ++ e)) + Right resp -> pure resp -- | 原样转发一条载荷给存储库 sendRawRequest :: BS.ByteString -> IO (Either String BS.ByteString) -sendRawRequest payload = do - current <- readMVar linkRef +sendRawRequest payload = withStorageLink $ \current -> do case current of Nothing -> pure (Left "storage not configured") Just link -> slRequest link payload @@ -212,10 +234,6 @@ doInsertMany env t rs = ask env "insert_batch" (ReqInsertBatch t rs) okOnly doDeleteKeys :: Env -> String -> [Int] -> IO (Either String ()) doDeleteKeys env t ks = ask env "delete_keys" (ReqDeleteKeys t ks) okOnly --- | 发 ReplaceAll -doReplaceAll :: Env -> String -> [Row] -> IO (Either String ()) -doReplaceAll env t rs = ask env "replace_all" (ReqReplaceAll t rs) okOnly - -- | 发 ListTables doListTables :: Env -> IO (Either String [String]) doListTables env = ask env "list_tables" ReqListTables $ \resp -> case resp of @@ -254,26 +272,6 @@ doDescribeTable env t = ask env "describe_table" (ReqDescribeTable t) $ \resp -> RespSchema info -> Just info _ -> Nothing --- | 给某一列建索引 -doCreateIndex :: Env -> String -> String -> IO (Either String ()) -doCreateIndex env t c = ask env "create_index" (ReqCreateIndex t c) okOnly - --- | 去掉某一列的索引 -doDropIndex :: Env -> String -> String -> IO (Either String ()) -doDropIndex env t c = ask env "drop_index" (ReqDropIndex t c) okOnly - --- | 删一列（列定义、这一列上的索引、每行里的那一格一起没） -doDropColumn :: Env -> String -> String -> IO (Either String ()) -doDropColumn env t c = ask env "drop_column" (ReqDropColumn t c) okOnly - --- | 发 ReplaceSchema（ALTER 用） -doReplaceSchema :: Env -> String -> [SchemaColumn] -> [Row] -> IO (Either String ()) -doReplaceSchema env t cols rs = ask env "replace_schema" (ReqReplaceSchema t cols rs) okOnly - --- | 发 CreateTable -doCreateTable :: Env -> String -> [SchemaColumn] -> IO (Either String ()) -doCreateTable env t cols = ask env "create_table" (ReqCreateTable t cols) okOnly - -- | 发 DropTable doDropTable :: Env -> String -> IO (Either String ()) doDropTable env t = ask env "drop_table" (ReqDropTable t) okOnly @@ -350,7 +348,7 @@ instance MonadStorage IPCStorage where deleteKeys t ks = IPCStorage (\env -> doDeleteKeys env t ks) -- \| 发 ReplaceAll - replaceAll t rs = IPCStorage (\env -> doReplaceAll env t rs) + replaceAll t rs = IPCStorage $ \env -> ask env "replace_all" (ReqReplaceAll t rs) okOnly -- \| 按某一列的索引取一行（没有索引就回 NoIndex） lookupByColumn t c k = IPCStorage (\env -> doLookupByColumn env t c k) @@ -359,45 +357,44 @@ instance MonadStorage IPCStorage where scanRange t c lo hi = IPCStorage (\env -> doScanRange env t c lo hi) -- \| 发 CreateTable - createTable name cols = IPCStorage (\env -> doCreateTable env name (map toWire cols)) + createTable name cols = IPCStorage $ \env -> ask env "create_table" (ReqCreateTable name (map toWire cols)) okOnly -- \| ALTER：整表换列定义与全部行 - replaceSchema name cols rows = IPCStorage (\env -> doReplaceSchema env name (map toWire cols) rows) + replaceSchema name cols rows = IPCStorage $ \env -> ask env "replace_schema" (ReqReplaceSchema name (map toWire cols) rows) okOnly -- \| 发 DropTable dropTable name = IPCStorage (\env -> doDropTable env name) -- \| 建索引 - createIndex t c = IPCStorage (\env -> doCreateIndex env t c) + createIndex t c = IPCStorage $ \env -> ask env "create_index" (ReqCreateIndex t c) okOnly -- \| 删索引 - dropIndex t c = IPCStorage (\env -> doDropIndex env t c) + dropIndex t c = IPCStorage $ \env -> ask env "drop_index" (ReqDropIndex t c) okOnly -- \| 删一列 - dropColumn t c = IPCStorage (\env -> doDropColumn env t c) + dropColumn t c = IPCStorage $ \env -> ask env "drop_column" (ReqDropColumn t c) okOnly -- \| 列表 + 逐表扫描拼库 snapshot = IPCStorage $ \env -> do result <- doListCatalog env case result of - Left _ -> pure [] - Right xs -> mapM (loadEntry env) xs + Left err -> pure (Left err) + Right xs -> fmap sequence (mapM (loadEntry env) xs) where + -- | 读取表数据并保留扫描错误 loadEntry env info = do rows <- doScan env (tiTable info) - pure + pure $ fmap (\rs -> ( tiTable info - , Table (tiTable info) (schemaToColumns (tiColumns info)) (either (const []) id rows) (Just (metaOf info)) - ) + , Table (tiTable info) (schemaToColumns (tiColumns info)) rs (Just (metaOf info)) + )) rows -- \| 只问数据字典要结构，设了当前库只留它的表 schema = IPCStorage $ \env -> do result <- ask env "all_catalogs" ReqAllCatalog $ \resp -> case resp of RespCatalog infos -> Just infos _ -> Nothing - pure $ case result of - Left _ -> [] - Right xs -> concatMap (visibleTables (envDatabase env)) xs + pure (fmap (concatMap (visibleTables (envDatabase env))) result) -- | 线上表信息转引擎侧统计 metaOf :: TableInfo -> TableMeta diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/Memory.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/Memory.hs index 2473fe6..7056ef8 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/Memory.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/Memory.hs @@ -73,10 +73,10 @@ instance MonadStorage MemoryStorage where ) -- \| 原样返回当前库 - snapshot = MemoryStorage $ \db -> Right (db, db) + snapshot = MemoryStorage $ \db -> Right (Right db, db) -- \| 返回实时行数与去重统计的无数据结构 - schema = MemoryStorage $ \db -> Right (map describe db, db) + schema = MemoryStorage $ \db -> Right (Right (map describe db), db) where -- | 将存储数据汇总为结构统计 describe (name, table) = diff --git a/chusql-core/engine/test/NormalizationSpec.hs b/chusql-core/engine/test/NormalizationSpec.hs index 1d004e4..a1e9e52 100644 --- a/chusql-core/engine/test/NormalizationSpec.hs +++ b/chusql-core/engine/test/NormalizationSpec.hs @@ -26,22 +26,20 @@ spec = do fmap (map verdict . findings) (analyzeNormalization schema (Declaration [["a", "b"]] [Dependency ["a"] ["c"]] True (Just True))) `shouldBe` Right [Satisfied, Violated, Violated, Violated] - it "detects transitive dependencies" $ do - fmap (map verdict . findings) (analyzeNormalization schema - (Declaration [["a"]] [Dependency ["b"] ["c"]] True (Just True))) + it "detects transitive dependencies and a preserving split" $ do + let analysis = analyzeNormalization schema + (Declaration [["a"]] [Dependency ["b"] ["c"]] True (Just True)) + fmap (map verdict . findings) analysis `shouldBe` Right [Satisfied, Satisfied, Violated, Violated] - it "distinguishes 3NF from BCNF using all candidate keys" $ do - fmap (map verdict . findings) (analyzeNormalization schema - (Declaration [["a", "b"]] [Dependency ["c"] ["b"]] True (Just True))) - `shouldBe` Right [Satisfied, Satisfied, Satisfied, Violated] - it "reports lossless decomposition that loses dependencies" $ do - fmap suggestions (analyzeNormalization schema - (Declaration [["a", "b"]] [Dependency ["c"] ["b"]] True (Just True))) - `shouldBe` Right [Decomposition [["b", "c"], ["a", "c"]] Satisfied Violated] - it "reports dependency preserving transitive split" $ do - fmap suggestions (analyzeNormalization schema - (Declaration [["a"]] [Dependency ["b"] ["c"]] True (Just True))) + fmap suggestions analysis `shouldBe` Right [Decomposition [["b", "c"], ["a", "b"]] Satisfied Satisfied] + it "distinguishes 3NF from BCNF and detects dependency loss" $ do + let analysis = analyzeNormalization schema + (Declaration [["a", "b"]] [Dependency ["c"] ["b"]] True (Just True)) + fmap (map verdict . findings) analysis + `shouldBe` Right [Satisfied, Satisfied, Satisfied, Violated] + fmap suggestions analysis + `shouldBe` Right [Decomposition [["b", "c"], ["a", "c"]] Satisfied Violated] it "does not certify incomplete dependencies" $ do fmap (map verdict . findings) (analyzeNormalization schema (Declaration [["a"]] [] False (Just True))) diff --git a/chusql-core/engine/test/RuntimeSpec.hs b/chusql-core/engine/test/RuntimeSpec.hs index 64b2836..85a237a 100644 --- a/chusql-core/engine/test/RuntimeSpec.hs +++ b/chusql-core/engine/test/RuntimeSpec.hs @@ -29,7 +29,6 @@ spec = do parseTypeExpr "MaybeInt" `shouldSatisfy` isFailure it "restores stable nested identities from bootstrap metadata" $ do let ty = tupleType [listType intType, maybeType stringType] - resolveType builtinTypes ty `shouldBe` resolveType builtinTypes ty fmap descriptorPhysical (resolveType builtinTypes ty) `shouldBe` Right (Product [Sequence SignedInteger, Optional Utf8]) it "distinguishes logical String from List Int" $ diff --git a/chusql-core/engine/test/Spec.hs b/chusql-core/engine/test/Spec.hs index 590a664..78526e3 100644 --- a/chusql-core/engine/test/Spec.hs +++ b/chusql-core/engine/test/Spec.hs @@ -17,17 +17,23 @@ import ChuSQL.Core.Model import ChuSQL.Core.Engine.Semantic (prepare) import ChuSQL.Core.Engine.Storage (IndexResult (..), MonadStorage (..)) import ChuSQL.Core.Engine.Storage.Memory (MemoryStorage (runMemoryStorage)) -import ChuSQL.Core.Engine.Storage.IPC (IPCStorage (..), closeConnection, defaultEnv, doListTables, envForDatabase, getDatabaseName, localStorageLink, runIPCStorage, setDatabaseName, setStorageLink) +import ChuSQL.Core.Engine.Storage.FFI (closeStorage, maintainStorage, openStorage) +import ChuSQL.Core.Engine.Storage.IPC (IPCStorage (..), StorageLink (..), closeConnection, defaultEnv, doListTables, envForDatabase, getDatabaseName, localStorageLink, runIPCStorage, sendRawRequest, setDatabaseName, setStorageLink) import ChuSQL.Core.Engine.Syntax.AST import ChuSQL.Core.Engine.Syntax.Parser -import Control.Concurrent (threadDelay) -import Control.Concurrent.MVar (MVar, modifyMVar_, newMVar, readMVar) -import Control.Exception (ErrorCall, IOException, finally, try) +import Control.Concurrent (forkFinally, forkIO, killThread, threadDelay) +import Control.Concurrent.Chan (newChan, readChan, writeChan) +import Control.Concurrent.MVar (MVar, modifyMVar_, newEmptyMVar, newMVar, putMVar, readMVar, takeMVar, tryPutMVar) +import Control.Monad (replicateM_, void) +import Control.Exception (ErrorCall, IOException, bracket_, finally, try) import Data.Hashable (hash) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as BSC import Data.List (isInfixOf, sortOn) import Data.Unique (hashUnique, newUnique) +import GHC.IO.Encoding (getForeignEncoding, latin1, setForeignEncoding) import System.CPUTime (getCPUTime) -import System.Directory (createDirectoryIfMissing, getTemporaryDirectory, removePathForcibly) +import System.Directory (createDirectoryIfMissing, getTemporaryDirectory, removePathForcibly, renameFile) import System.FilePath (()) import System.Timeout (timeout) import Test.Hspec @@ -118,6 +124,81 @@ domainSql db (sql : rest) = do -- | 跑引擎层全部 hspec 用例 main :: IO () main = hspec $ do + describe "Storage read failures" $ do + it "propagates catalog failures through snapshot, schema and SQL" $ do + setStorageLink StorageLink + { slRequest = \_ -> pure (Right (BSC.pack "{\"status\":\"error\",\"message\":\"catalog unavailable\"}")) + , slClose = pure () + } + flip finally closeConnection $ do + fmap (fmap (map fst)) (runIPCStorage snapshot) `shouldReturn` Left "catalog unavailable" + fmap (fmap (map fst)) (runIPCStorage schema) `shouldReturn` Left "catalog unavailable" + runIPCStorage (runStatementM (CreateTable "blocked" [("id", TInt)])) + `shouldReturn` Left "catalog unavailable" + + it "rejects the entire snapshot when a table scan fails" $ do + setStorageLink StorageLink + { slRequest = \payload -> pure $ Right $ BSC.pack $ + if BSC.pack "list_catalog" `BS.isInfixOf` payload + then "{\"status\":\"catalog\",\"schemas\":[{\"table\":\"broken\",\"columns\":[],\"row_count\":1}]}" + else "{\"status\":\"error\",\"message\":\"data page unreadable\"}" + , slClose = pure () + } + (fmap (fmap (map fst)) (runIPCStorage snapshot) `shouldReturn` Left "data page unreadable") `finally` closeConnection + + describe "Storage lifecycle and cancellation" $ do + it "waits for concurrent requests before closing the storage link" $ do + started <- newChan + release <- newEmptyMVar + finished <- newChan + closing <- newEmptyMVar + closed <- newEmptyMVar + setStorageLink StorageLink + { slRequest = \_ -> writeChan started () >> readMVar release >> pure (Right BS.empty) + , slClose = putMVar closed () + } + flip finally (void (tryPutMVar release ()) >> closeConnection) $ do + replicateM_ 2 (void (forkIO (sendRawRequest BS.empty >>= writeChan finished))) + timeout 1000000 (replicateM_ 2 (readChan started)) `shouldReturn` Just () + void (forkIO (putMVar closing () >> closeConnection)) + takeMVar closing + timeout 100000 (readMVar closed) `shouldReturn` Nothing + putMVar release () + replicateM_ 2 (timeout 1000000 (readChan finished) `shouldReturn` Just (Right BS.empty)) + timeout 1000000 (readMVar closed) `shouldReturn` Just () + sendRawRequest BS.empty `shouldReturn` Left "storage not configured" + closeConnection + + it "keeps the pool usable after cancellation while all workers are busy" $ do + pool <- workerPool 4 + started <- newChan + release <- newEmptyMVar + finished <- newEmptyMVar + void (forkIO (poolRun pool (replicate 4 (writeChan started () >> readMVar release)) >> putMVar finished ())) + flip finally (void (tryPutMVar release ())) $ do + timeout 1000000 (replicateM_ 4 (readChan started)) `shouldReturn` Just () + replicateM_ 2 (timeout 100000 (poolRun pool (replicate 3 (pure ()))) `shouldReturn` Nothing) + putMVar release () + timeout 1000000 (takeMVar finished) `shouldReturn` Just () + timeout 1000000 (poolRun pool [pure ()]) `shouldReturn` Just [()] + + it "releases the storage link when an in-flight request is cancelled" $ do + started <- newEmptyMVar + release <- newEmptyMVar + finished <- newEmptyMVar + closed <- newEmptyMVar + setStorageLink StorageLink + { slRequest = \_ -> putMVar started () >> takeMVar release >> pure (Right BS.empty) + , slClose = putMVar closed () + } + thread <- forkFinally (sendRawRequest BS.empty) (const (putMVar finished ())) + flip finally (killThread thread >> closeConnection) $ do + timeout 1000000 (takeMVar started) `shouldReturn` Just () + killThread thread + timeout 1000000 (takeMVar finished) `shouldReturn` Just () + timeout 1000000 closeConnection `shouldReturn` Just () + readMVar closed `shouldReturn` () + NormalizationSpec.spec RuntimeSpec.spec describe "identity syntax" $ do @@ -1772,9 +1853,10 @@ main = hspec $ do sameResultAsUnoptimized "SELECT d.name FROM (SELECT name, age FROM users) d WHERE d.age > 20 ORDER BY d.name" it "supplies memory schema statistics without exposing rows" $ do - case runMemoryStorage (schema :: MemoryStorage Database) testDB of + case runMemoryStorage schema testDB of Left err -> expectationFailure err - Right (structures, _) -> do + Right (Left err, _) -> expectationFailure err + Right (Right structures, _) -> do map (tableRows . snd) structures `shouldBe` [[], []] fmap (fmap metaRowCount . tableMeta) (lookup "users" structures) `shouldBe` Just (Just 3) @@ -2094,6 +2176,51 @@ main = hspec $ do ] describe "ChuSQL.Core.Engine.Storage.IPC" $ do + it "uses UTF-8 config paths regardless of the foreign string encoding" $ do + tmp <- getTemporaryDirectory + stamp <- uniqueStamp + let dir = tmp ("chusql-utf8-" ++ stamp) + config = dir "settings-é.toml" + createDirectoryIfMissing True dir + flip finally (cleanSessionDir dir) $ do + original <- writeSessionConfig dir + renameFile original config + encoding <- getForeignEncoding + bracket_ (setForeignEncoding latin1) (setForeignEncoding encoding) $ do + opened <- openStorage (Just config) + case opened of + Left err -> expectationFailure err + Right handle -> closeStorage handle + maintainStorage config (BSC.pack "{\"mode\":\"unknown\"}") `shouldReturn` Left "invalid maintenance mode" + + it "preserves each worker's storage open error" $ do + tmp <- getTemporaryDirectory + stamp <- uniqueStamp + pool <- workerPool 4 + let paths = [tmp ("missing-ffi-" ++ stamp ++ "-" ++ show n) | n <- [1 .. 16 :: Int]] + -- | 检查打开失败并归还意外成功的句柄 + checkOpen path = do + opened <- openStorage (Just path) + case opened of + Left err -> do + err `shouldContain` "config file not found" + err `shouldContain` path + Right handle -> closeStorage handle >> expectationFailure "expected missing config error" + poolRun pool (map checkOpen paths) >> pure () + + it "returns FFI input errors from workers and remains usable" $ do + withTestSession $ \srv -> do + pool <- workerPool 4 + -- | 检查无效 UTF-8 的请求与维护错误 + let checkErrors = do + requested <- sendRawRequest (BS.pack [255]) + requested `shouldSatisfy` either (isInfixOf "request is not valid UTF-8") (const False) + maintained <- maintainStorage (dataDir srv ++ ".toml") (BS.pack [255]) + maintained `shouldSatisfy` either (isInfixOf "invalid utf-8") (const False) + poolRun pool (replicate 16 checkErrors) >> pure () + sendRawRequest (BSC.pack "{\"method\":\"ping\"}") + `shouldReturn` Right (BSC.pack "{\"status\":\"pong\"}") + it "persists runtime type identities and nested values across restart" $ do withTestSession $ \srv -> do -- | 解析并通过真实存储执行 SQL @@ -2143,42 +2270,21 @@ main = hspec $ do query "SELECT secret FROM reused" `shouldReturn` Right [[("secret", VStr "new")]] query "SELECT secret FROM reused WHERE id = 1" `shouldReturn` Right [[("secret", VStr "new")]] - it "scan returns rows with all three value types after insert" $ do + it "scans stored value types and reports missing tables" $ do withTestSession $ \srv -> withServerEnv srv $ do let row = [("id", VInt 7), ("name", VStr "Alice"), ("flag", VBool True)] runIPCStorage (insert "users" row) `shouldReturn` Right () rows <- runIPCStorage (scan "users") (map sortRow <$> rows) `shouldBe` Right [sortRow row] - - it "scan on a missing table returns Left" $ do - withTestSession $ \srv -> withServerEnv srv $ do result <- runIPCStorage (scan "no_such_table") result `shouldSatisfy` isLeft it "scan returns all 20 inserted rows in order" $ do withTestSession $ \srv -> withServerEnv srv $ do - mapM_ (\i -> runIPCStorage (insert "many" [("id", VInt i)])) [1 .. 20 :: Int] + runIPCStorage (insertMany "many" [[("id", VInt i)] | i <- [1 .. 20 :: Int]]) `shouldReturn` Right () rows <- runIPCStorage (scan "many") fmap length rows `shouldBe` Right 20 - fmap (sortOn show . map (lookup "id")) rows - `shouldBe` Right (sortOn show (map (Just . VInt) [1 .. 20])) - - it "replaceAll leaves only the new rows" $ do - withTestSession $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "t" [("id", VInt 1)]) `shouldReturn` Right () - runIPCStorage (replaceAll "t" [[("id", VInt 9)], [("id", VInt 8)]]) `shouldReturn` Right () - rows <- runIPCStorage (scan "t") - fmap (sortOn show . map (lookup "id")) rows - `shouldBe` Right (sortOn show [Just (VInt 9), Just (VInt 8)]) - - it "lookupByColumn finds a row inserted with an id" $ do - withTestSession $ \srv -> withServerEnv srv $ do - let row = [("id", VInt 7), ("name", VStr "Zoe")] - runIPCStorage (insert "keyed" row) `shouldReturn` Right () - found <- runIPCStorage (lookupByColumn "keyed" "id" (VInt 7)) - fmap sortRow (indexRow found) `shouldBe` Just (sortRow row) - missing <- runIPCStorage (lookupByColumn "keyed" "id" (VInt 8)) - missing `shouldBe` Right (IndexRows []) + fmap (map (lookup "id")) rows `shouldBe` Right (map (Just . VInt) [1 .. 20]) it "looks up a string index" $ do withTestSession $ \srv -> withServerEnv srv $ do @@ -2202,21 +2308,27 @@ main = hspec $ do fmap (fmap (map (lookup "id"))) open `shouldBe` Right (Just [Just (VInt 3)]) runIPCStorage (scanRange "rng" "nope" Nothing Nothing) `shouldReturn` Right Nothing - it "replaceAll rebuilds the index" $ do + it "looks up inserted ids and replaces rows with rebuilt indexes" $ do withTestSession $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "rb" [("id", VInt 1)]) `shouldReturn` Right () - runIPCStorage (replaceAll "rb" [[("id", VInt 9), ("name", VStr "Zoe")]]) `shouldReturn` Right () + let original = [("id", VInt 1), ("name", VStr "Alice")] + replacement = [("id", VInt 9), ("name", VStr "Zoe")] + runIPCStorage (insert "rb" original) `shouldReturn` Right () + found <- runIPCStorage (lookupByColumn "rb" "id" (VInt 1)) + fmap sortRow (indexRow found) `shouldBe` Just (sortRow original) + runIPCStorage (lookupByColumn "rb" "id" (VInt 8)) `shouldReturn` Right (IndexRows []) + runIPCStorage (replaceAll "rb" [replacement, [("id", VInt 8)]]) `shouldReturn` Right () + rows <- runIPCStorage (scan "rb") + fmap (map sortRow) rows `shouldBe` Right [sortRow replacement, [("id", VInt 8)]] old <- runIPCStorage (lookupByColumn "rb" "id" (VInt 1)) old `shouldBe` Right (IndexRows []) new <- runIPCStorage (lookupByColumn "rb" "id" (VInt 9)) - fmap sortRow (indexRow new) - `shouldBe` Just (sortRow [("id", VInt 9), ("name", VStr "Zoe")]) + fmap sortRow (indexRow new) `shouldBe` Just (sortRow replacement) it "snapshot lists tables and scans each one to build the database" $ do withTestSession $ \srv -> withServerEnv srv $ do runIPCStorage (insert "a" [("id", VInt 1)]) `shouldReturn` Right () runIPCStorage (insert "b" [("id", VInt 2)]) `shouldReturn` Right () - db <- runIPCStorage snapshot + Right db <- runIPCStorage snapshot map fst db `shouldBe` ["a", "b"] map (length . tableRows . snd) db `shouldBe` [1, 1] map (map (lookup "id") . tableRows . snd) db `shouldBe` [[Just (VInt 1)], [Just (VInt 2)]] @@ -2346,7 +2458,7 @@ main = hspec $ do (Insert "meta_t" ["id", "code"] [[LitInt 1, LitInt 500], [LitInt 2, LitInt 501]]) ) _ <- runIPCStorage (runStatementM (CreateIndex "meta_t" "code")) - db <- runIPCStorage schema + Right db <- runIPCStorage schema case lookup "meta_t" db of Nothing -> expectationFailure "meta_t missing from schema" Just t -> do @@ -2857,6 +2969,19 @@ main = hspec $ do errorCategory "unsupported protocol version" `shouldBe` ProtocolError errorCategory "something else" `shouldBe` UnknownError + describe "Worker pool capacity" $ do + it "grows the existing queue so old references use the additional workers" $ do + original <- workerPool 4 + _ <- workerPool 8 + started <- newChan + release <- newEmptyMVar + finished <- newEmptyMVar + void (forkIO (poolRun original (replicate 8 (writeChan started () >> readMVar release)) >> putMVar finished ())) + flip finally (void (tryPutMVar release ())) $ do + timeout 1000000 (replicateM_ 8 (readChan started)) `shouldReturn` Just () + putMVar release () + timeout 1000000 (takeMVar finished) `shouldReturn` Just () + -- | 每个测试表都清空行后的库 emptyDB :: Database emptyDB = [(n, t{tableRows = []}) | (n, t) <- testDB] @@ -2912,7 +3037,7 @@ cleanSessionDir dir = do _ <- try (removePathForcibly (dir ++ ".toml")) :: IO (Either IOException ()) pure () --- | 一个用例一份干净数据目录；动态库打不开则 pending +-- | 每例独立目录，存储打开失败则测试失败 withTestSession :: (StorageSession -> IO ()) -> IO () withTestSession act = do tmp <- getTemporaryDirectory @@ -2921,7 +3046,7 @@ withTestSession act = do createDirectoryIfMissing True dir opened <- openSessionAt dir case opened of - Left err -> cleanSessionDir dir >> pendingWith err + Left err -> cleanSessionDir dir >> expectationFailure err Right session -> act session `finally` (closeConnection >> cleanSessionDir dir) -- | 语句不带库名，先建好测试库再切进去 @@ -3050,7 +3175,7 @@ instance MonadStorage ProbeStorage where dropTable _ = ProbeStorage (\_ -> pure (Left "probe: dropTable is not supported")) dropColumn _ _ = ProbeStorage (\_ -> pure (Left "probe: dropColumn is not supported")) replaceSchema _ _ _ = ProbeStorage (\_ -> pure (Left "probe: replaceSchema is not supported")) - snapshot = ProbeStorage (\_ -> pure []) + snapshot = ProbeStorage (\_ -> pure (Right [])) -- | 判断 Either 是 Right isRight :: Either a b -> Bool @@ -3271,7 +3396,7 @@ instance MonadStorage Counting where -- \| 换结构：走内存实现 replaceSchema name cols rows = memory (memReplaceSchema name cols rows) -- \| 快照：把当前库原样交出去 - snapshot = Counting $ \db -> Right (db, db, []) + snapshot = Counting $ \db -> Right (Right db, db, []) -- | 解析一条 SQL 并在计数存储上跑，返回结果与扫描过的表 countedRun :: String -> (Either String [Row], [String]) diff --git a/chusql-core/storage/src/btree.rs b/chusql-core/storage/src/btree.rs index 20295ac..d070656 100644 --- a/chusql-core/storage/src/btree.rs +++ b/chusql-core/storage/src/btree.rs @@ -167,23 +167,47 @@ fn put_node(p: &mut Page, node: &Node) -> io::Result<()> { } /// 从一页读出节点 -fn get_node(p: &Page) -> Node { +fn get_node(p: &Page) -> io::Result { + if p.data.len() < HEADER { + return Err(io::Error::new(io::ErrorKind::InvalidData, "truncated index node header")); + } let kind = read_u8(p, OFF_KIND); let count = read_u16(p, OFF_COUNT) as usize; + let slots_end = HEADER + count * SLOT; + if !matches!(kind, NODE_LEAF | NODE_INTERNAL) || slots_end > p.data.len() { + return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid index node kind or slot count")); + } let trailer = read_u64(p, OFF_NEXT); let mut keys = Vec::with_capacity(count); let mut values = Vec::with_capacity(count + 1); + let mut cell_end = p.data.len(); for i in 0..count { let off = read_u16(p, HEADER + i * SLOT) as usize; + if off < slots_end || off + CELL_FIXED > p.data.len() { + return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid index cell offset")); + } let klen = read_u16(p, off) as usize; + if klen > MAX_KEY_BYTES || off + CELL_FIXED + klen > p.data.len() { + return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid index key length")); + } + if off + CELL_FIXED + klen > cell_end { + return Err(io::Error::new(io::ErrorKind::InvalidData, "overlapping index cells")); + } + cell_end = off; keys.push(p.data[off + 2..off + 2 + klen].to_vec()); values.push(read_u64(p, off + 2 + klen)); } + if keys.windows(2).any(|pair| pair[0] > pair[1]) { + return Err(io::Error::new(io::ErrorKind::InvalidData, "index keys are not sorted")); + } if kind == NODE_INTERNAL { values.push(trailer); - Node { kind, next: 0, keys, values } + if values.contains(&META_PAGE) { + return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid index child page")); + } + Ok(Node { kind, next: 0, keys, values }) } else { - Node { kind, next: trailer, keys, values } + Ok(Node { kind, next: trailer, keys, values }) } } @@ -299,7 +323,7 @@ impl DiskBTree { /// 按页号读节点 fn load(&mut self, id: PageId) -> io::Result { - self.file.with_page(id, get_node) + self.file.with_page(id, get_node)? } /// 按页号写节点 @@ -338,24 +362,9 @@ impl DiskBTree { /// 按 key 取全部位置（重复键按值升序） pub fn get_all(&mut self, key: &[u8]) -> io::Result> { - let Some(mut id) = self.root()? else { - return Ok(Vec::new()); - }; - loop { - let node = self.load(id)?; - if node.kind == NODE_LEAF { - let start = node.keys.partition_point(|k| k.as_slice() < key); - let mut out = Vec::new(); - for i in start..node.keys.len() { - if node.keys[i].as_slice() != key { - break; - } - out.push(node.values[i]); - } - return Ok(out); - } - id = node.values[child_slot(&node.keys, key)]; - } + let mut values: Vec<_> = self.scan(Some(key), Some(key))?.into_iter().map(|(_, value)| value).collect(); + values.sort_unstable(); + Ok(values) } /// 插入一个键值对（键可重复） @@ -412,10 +421,9 @@ impl DiskBTree { self.store(id, &node)?; return Ok(None); } - let (sep, mut right) = split_node(&mut node); + let (sep, right) = split_node(&mut node); let right_id = self.alloc(&right)?; if node.kind == NODE_LEAF { - right.next = node.next; node.next = right_id; } self.store(id, &node)?; @@ -443,8 +451,14 @@ impl DiskBTree { } return Ok(false); } - let child = node.values[child_slot(&node.keys, key)]; - self.delete_rec(child, key, value) + let first = node.keys.partition_point(|k| k.as_slice() < key); + let last = child_slot(&node.keys, key); + for &child in &node.values[first..=last] { + if self.delete_rec(child, key, value)? { + return Ok(true); + } + } + Ok(false) } /// 叶子链上按顺序取全部 (键, 值) @@ -475,7 +489,7 @@ impl DiskBTree { } id = match lo { None => node.values[0], - Some(start) => node.values[child_slot(&node.keys, start)], + Some(start) => node.values[node.keys.partition_point(|k| k.as_slice() < start)], }; } let mut out = Vec::new(); diff --git a/chusql-core/storage/src/config.rs b/chusql-core/storage/src/config.rs index c5177e4..7990170 100644 --- a/chusql-core/storage/src/config.rs +++ b/chusql-core/storage/src/config.rs @@ -229,17 +229,17 @@ pub fn resolve(text: Option<&str>, path: Option) -> Result, path: Option) -> Result Result<(), String> { } Ok(()) } - -#[cfg(test)] -mod tests { - use super::*; - - /// 别的层的分区不能被当成错误 - #[test] - fn ignores_sections_owned_by_other_layers() { - let text = r#" -[page] -size = 8192 - -[storage] -data_dir = "../data" - -[server] -listen_host = "127.0.0.1" -port = 7777 -max_message = 1048576 - -[web] -port = 7778 -"#; - let loaded = resolve(Some(text), Some(PathBuf::from("settings.toml"))).unwrap(); - assert_eq!(loaded.config.page_size, 8192); - assert_eq!(loaded.config.data_dir, PathBuf::from("../data")); - } - - /// 自己分区里的拼写错误仍然要报错 - #[test] - fn still_rejects_typos_inside_its_own_sections() { - let text = "[page]\npage_size = 8192\n"; - assert!(resolve(Some(text), None).is_err()); - } -} diff --git a/chusql-core/storage/src/runtime.rs b/chusql-core/storage/src/runtime.rs index a0d4b0e..cfdeaf1 100644 --- a/chusql-core/storage/src/runtime.rs +++ b/chusql-core/storage/src/runtime.rs @@ -1021,17 +1021,12 @@ impl Server { } } - /// 系统目录里账号表已登记 - fn system_initialized(&self) -> bool { - self.catalog - .lock() - .map(|c| c.describe(USERS_TABLE).is_some_and(|schema| schema.system)) - .unwrap_or(false) - } - - /// 最近一次落盘的检查点 LSN - fn last_checkpoint(&self) -> u64 { - self.wal.read_checkpoint().ok().flatten().unwrap_or(0) + /// 读取系统初始化状态与检查点 LSN + fn system_status(&self) -> std::io::Result { + let initialized = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))? + .describe(USERS_TABLE).is_some_and(|schema| schema.system); + let last_lsn = self.wal.read_checkpoint()?.unwrap_or(0); + Ok(Response::System { initialized, last_lsn }) } /// 读账号表；表没登记或数据文件缺失都算错 @@ -1346,12 +1341,12 @@ impl Server { } } Request::BootstrapSystem { user, password_hash } => { - match self.bootstrap_system(user.as_deref(), password_hash.as_deref()) { - Ok(_) => Response::System { initialized: true, last_lsn: self.last_checkpoint() }, + match self.bootstrap_system(user.as_deref(), password_hash.as_deref()).and_then(|_| self.system_status()) { + Ok(response) => response, Err(e) => Response::Error { message: e.to_string() }, } } - Request::SystemStatus => Response::System { initialized: self.system_initialized(), last_lsn: self.last_checkpoint() }, + Request::SystemStatus => self.system_status().unwrap_or_else(|e| Response::Error { message: e.to_string() }), Request::Ping => { log_debug!(request, "ping"); Response::Pong diff --git a/chusql-core/storage/tests/btree_test.rs b/chusql-core/storage/tests/btree_test.rs index ca9aecf..7796e5f 100644 --- a/chusql-core/storage/tests/btree_test.rs +++ b/chusql-core/storage/tests/btree_test.rs @@ -1,9 +1,45 @@ use chusql_core_storage::btree::{fits_in_page, DiskBTree, MAX_KEY_BYTES}; use chusql_core_storage::config::{DEFAULT_BTREE_ORDER, DEFAULT_PAGE_SIZE}; use chusql_core_storage::page::DEFAULT_POOL_SIZE; +use std::io::{ErrorKind, Seek, SeekFrom, Write}; // B+ 树测试：字节键、重复键、范围扫描、重开与配置校验。 +/// 损坏节点返回数据错误且重复读取不 panic +#[test] +fn malformed_node_fields_return_invalid_data() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("corrupt.idx"); + { + let mut tree = open(dir.path(), "corrupt.idx"); + tree.insert(b"x", 1).unwrap(); + tree.insert(b"y", 2).unwrap(); + } + let original = std::fs::read(&path).unwrap(); + let cell = u16::from_le_bytes([original[DEFAULT_PAGE_SIZE + 11], original[DEFAULT_PAGE_SIZE + 12]]) as usize; + for (name, offset, bytes) in [ + ("slot count", 1, vec![255, 255]), + ("node kind", 0, vec![2]), + ("cell overlaps slots", 11, vec![0, 0]), + ("cell beyond page", 11, vec![255, 15]), + ("overlapping cells", 13, (cell as u16).to_le_bytes().to_vec()), + ("key length", cell, vec![255, 255]), + ("key order", cell + 2, vec![b'z']), + ("zero child", 0, vec![1]), + ] { + std::fs::write(&path, &original).unwrap(); + let mut file = std::fs::OpenOptions::new().write(true).open(&path).unwrap(); + file.seek(SeekFrom::Start((DEFAULT_PAGE_SIZE + offset) as u64)).unwrap(); + file.write_all(&bytes).unwrap(); + drop(file); + let mut tree = open(dir.path(), "corrupt.idx"); + for _ in 0..2 { + let error = tree.get(b"x").unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidData, "{name}: {error}"); + } + } +} + /// 按默认参数打开指定索引文件 fn open(dir: &std::path::Path, name: &str) -> DiskBTree { DiskBTree::open(dir.join(name), DEFAULT_PAGE_SIZE, DEFAULT_BTREE_ORDER, DEFAULT_POOL_SIZE).unwrap() @@ -14,6 +50,29 @@ fn key(n: i64) -> [u8; 8] { (n as u64 ^ (1u64 << 63)).to_be_bytes() } +/// 重复键跨页后仍能完整查找和删除 +#[test] +fn duplicate_keys_across_splits() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("split-duplicates.idx"); + let mut tree = DiskBTree::open(&path, DEFAULT_PAGE_SIZE, 4, DEFAULT_POOL_SIZE).unwrap(); + for value in (0..100).rev() { + tree.insert(b"same", value).unwrap(); + } + drop(tree); + let mut tree = DiskBTree::open(&path, DEFAULT_PAGE_SIZE, 4, DEFAULT_POOL_SIZE).unwrap(); + assert_eq!(tree.get(b"same").unwrap(), Some(0)); + assert_eq!(tree.get_all(b"same").unwrap(), (0..100).collect::>()); + assert_eq!(tree.scan(Some(b"same"), Some(b"same")).unwrap().len(), 100); + for value in 0..100 { + assert!(tree.delete(b"same", value).unwrap()); + } + assert!(!tree.delete(b"same", 0).unwrap()); + drop(tree); + let mut tree = DiskBTree::open(&path, DEFAULT_PAGE_SIZE, 4, DEFAULT_POOL_SIZE).unwrap(); + assert!(tree.get_all(b"same").unwrap().is_empty()); +} + /// 空树查不到，遍历为空 #[test] fn empty_tree_get() { diff --git a/chusql-core/storage/tests/config_test.rs b/chusql-core/storage/tests/config_test.rs index a2dce69..4cf93c5 100644 --- a/chusql-core/storage/tests/config_test.rs +++ b/chusql-core/storage/tests/config_test.rs @@ -187,13 +187,14 @@ fn load_without_explicit_path_uses_builtin_defaults() { } } -/// [server] 分区归别的层管；本层拼写错误要报错 +/// 忽略其他层分区并拒绝存储配置拼写错误 #[test] fn server_section_belongs_to_another_layer() { let _guard = env_guard(); - let db_server = "[server]\nlisten_host = \"0.0.0.0\"\nport = 7778\nmax_rows = 10\n"; + let db_server = "[page]\nsize = 8192\n[storage]\ndata_dir = \"../data\"\n[server]\nlisten_host = \"0.0.0.0\"\nport = 7778\nmax_rows = 10\n[web]\nport = 7778\n"; let loaded = config::resolve(Some(db_server), None).unwrap(); - assert_eq!(loaded.config.data_dir, config::default_data_dir()); + assert_eq!(loaded.config.page_size, 8192); + assert_eq!(loaded.config.data_dir, PathBuf::from("../data")); let typo = "[storage]\ndata_dirs = \"oops\"\n"; assert!(config::resolve(Some(typo), None).is_err()); diff --git a/chusql-core/storage/tests/runtime_test.rs b/chusql-core/storage/tests/runtime_test.rs index 4b1bfbf..556e498 100644 --- a/chusql-core/storage/tests/runtime_test.rs +++ b/chusql-core/storage/tests/runtime_test.rs @@ -5,6 +5,38 @@ use chusql_core_storage::runtime::Storage; // 进程内存储集成测试：协议请求、索引、账号与配置。 +/// 检查点损坏不冒充零 LSN且可恢复读取 +#[test] +fn system_status_reports_checkpoint_read_errors() -> Result<(), Box> { + let (_server, data) = start_server(); + let mut connection = connect()?; + let checkpoint = data.path().join("system/wal.checkpoint"); + let initial: serde_json::Value = serde_json::from_str(&send_raw(&mut connection, r#"{"method":"system_status"}"#))?; + assert_eq!(initial["last_lsn"], 0); + request_ok(&mut connection, serde_json::json!({"method":"object_acl_replace","expected":null,"payload":"{}"}))?; + let original = std::fs::read(&checkpoint)?; + let before: serde_json::Value = serde_json::from_str(&send_raw(&mut connection, r#"{"method":"system_status"}"#))?; + assert_eq!(before["status"], "system"); + assert!(before["last_lsn"].as_u64().unwrap() > 0); + std::fs::write(&checkpoint, b"bad")?; + let broken: serde_json::Value = serde_json::from_str(&send_raw(&mut connection, r#"{"method":"system_status"}"#))?; + assert_eq!(broken["status"], "error"); + assert!(broken["message"].as_str().unwrap().contains("invalid WAL checkpoint length")); + assert!(broken.get("last_lsn").is_none()); + assert!(send_raw(&mut connection, r#"{"method":"ping"}"#).contains("pong")); + std::fs::write(&checkpoint, &original)?; + let restored: serde_json::Value = serde_json::from_str(&send_raw(&mut connection, r#"{"method":"system_status"}"#))?; + assert_eq!(restored, before); + std::fs::remove_file(&checkpoint)?; + std::fs::create_dir(&checkpoint)?; + let unreadable: serde_json::Value = serde_json::from_str(&send_raw(&mut connection, r#"{"method":"system_status"}"#))?; + assert_eq!(unreadable["status"], "error"); + assert!(unreadable.get("last_lsn").is_none()); + std::fs::remove_dir(&checkpoint)?; + std::fs::write(&checkpoint, original)?; + Ok(()) +} + /// 权限快照支持跨页文本并拒绝旧版本覆盖 #[test] fn object_acl_snapshots_are_private_and_compare_versions() -> Result<(), Box> { diff --git a/chusql-server/src/ChuSQL/Server/Backend.hs b/chusql-server/src/ChuSQL/Server/Backend.hs index 4a0fcc1..59ad535 100644 --- a/chusql-server/src/ChuSQL/Server/Backend.hs +++ b/chusql-server/src/ChuSQL/Server/Backend.hs @@ -108,7 +108,7 @@ ipcBackend = do RespError err -> Left err _ -> Left "unexpected database response" , beStorage = \payload -> runRaw payload - , beSnapshot = Right <$> runIPCStorageIn (snapshot :: IPCStorage Database) (sessionEnv current) + , beSnapshot = runIPCStorageIn snapshot (sessionEnv current) , beApplyTransaction = \ops -> do response <- sendRequest (ReqInDatabase current (ReqApplyTransaction ops)) pure $ case response of @@ -148,10 +148,10 @@ ipcBackend = do Right rows -> do cols <- case rows of [] -> do - db <- runIPCStorageIn (schema :: IPCStorage Database) env - pure (columnsFromStatement db stmt) - (r : _) -> pure (map fst r) - pure (Right (StatementResult cols rows)) + db <- runIPCStorageIn schema env + pure (fmap (\structures -> columnsFromStatement structures stmt) db) + (r : _) -> pure (Right (map fst r)) + pure (fmap (\names -> StatementResult names rows) cols) -- | 测试用内存后端，状态是一个 Database memoryBackend :: String -> MVar Database -> Backend diff --git a/chusql-server/src/ChuSQL/Server/TCP.hs b/chusql-server/src/ChuSQL/Server/TCP.hs index 0401484..2994f89 100644 --- a/chusql-server/src/ChuSQL/Server/TCP.hs +++ b/chusql-server/src/ChuSQL/Server/TCP.hs @@ -59,8 +59,8 @@ import Network.Socket ( socket, socketToHandle, ) -import System.IO (BufferMode (LineBuffering), IOMode (ReadWriteMode), hClose, hFlush, hPutStrLn, hSetBuffering, hSetNewlineMode, noNewlineTranslation, stderr) -import System.IO.Error (isEOFError) +import System.IO (BufferMode (LineBuffering), Handle, IOMode (ReadWriteMode), hClose, hFlush, hPutStrLn, hSetBuffering, hSetNewlineMode, noNewlineTranslation, stderr) +import System.IO.Error (eofErrorType, isEOFError, mkIOError) -- TCP 服务器：监听端口，一连接一会话，按一行一 JSON 收发协议消息。 @@ -255,20 +255,19 @@ serveConnection env conn = do entry <- newLiveEntry (isLocalPeer peer) unregister <- registerConnection (srvLive env) entry session <- newSessionWith (srvBackend env) (srvRootName env) (srvSettingsFile env) (\names -> void (dropConnectionsOf (srvLive env) names)) - outcome <- try (loop h session entry `finally` unregister) :: IO (Either IOException ()) + outcome <- try (loop h session entry BSC.empty `finally` unregister) :: IO (Either IOException ()) case outcome of Left err | not (isEOFError err) -> hPutStrLn stderr ("connection error: " ++ show err) _ -> pure () void (try (hClose h) :: IO (Either IOException ())) where -- | 逐行读请求并应答 - loop h session entry = do - line <- BSC.hGetLine h - let cleaned = BSC.dropWhileEnd (== '\r') line - if BSC.length cleaned > scMaxMessage (srvConfig env) - then send h (RespError "too_large" "request line too large") >> loop h session entry - else case decodeRequest cleaned of - Left message -> send h (RespError "bad_request" message) >> loop h session entry + loop h session entry pending = do + frame <- readRequestLine h (scMaxMessage (srvConfig env)) pending + case frame of + Nothing -> send h (RespError "too_large" "request line too large") + Just (cleaned, remaining) -> case decodeRequest cleaned of + Left message -> send h (RespError "bad_request" message) >> loop h session entry remaining Right request -> do expired <- readIORef (leDropped entry) if expired @@ -276,13 +275,36 @@ serveConnection env conn = do else do (response, keepGoing) <- dispatch env session entry request send h response - if keepGoing then loop h session entry else pure () + if keepGoing then loop h session entry remaining else pure () -- | 发一条响应并冲缓冲 send h response = do BLC.hPutStr h (encodeResponse response) BLC.hPutStr h "\n" hFlush h +-- | 限长读取一行并保留后续帧 +readRequestLine :: Handle -> Int -> BSC.ByteString -> IO (Maybe (BSC.ByteString, BSC.ByteString)) +readRequestLine handle limit = collect [] 0 + where + -- | 按块读取至换行、超限或 EOF + collect parts total bytes = case BSC.elemIndex '\n' bytes of + Just offset -> finish (BSC.take offset bytes : parts) (BSC.drop (offset + 1) bytes) + Nothing + | used > limit && (used - limit > 1 || BSC.null bytes || BSC.last bytes /= '\r') -> pure Nothing + | otherwise -> do + next <- BSC.hGetSome handle (max 1 (min 4094 (limit - used) + 2)) + if BSC.null next + then if used == 0 + then ioError (mkIOError eofErrorType "read request" (Just handle) Nothing) + else finish (bytes : parts) BSC.empty + else collect (bytes : parts) used next + where + used = total + BSC.length bytes + -- | 合并并校验一行的有效长度 + finish parts remaining = do + let line = BSC.dropWhileEnd (== '\r') (BSC.concat (reverse parts)) + pure (if BSC.length line > limit then Nothing else Just (line, remaining)) + -- | 一条请求一条响应，quit 后停连接 dispatch :: ServerEnv -> Session -> LiveEntry -> ClientRequest -> IO (ServerResponse, Bool) dispatch env session entry request = case request of diff --git a/chusql-server/test/Spec.hs b/chusql-server/test/Spec.hs index a18bf6a..c2df1dd 100644 --- a/chusql-server/test/Spec.hs +++ b/chusql-server/test/Spec.hs @@ -51,7 +51,7 @@ import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar (MVar, modifyMVar_, newEmptyMVar, newMVar, putMVar, takeMVar) import System.Timeout (timeout) import Control.Exception (IOException, bracket, finally, try) -import Control.Monad (void) +import Control.Monad (replicateM_, void) import Data.List (sort) import Data.Aeson (decode) import qualified Data.Aeson as A @@ -729,6 +729,24 @@ tcpServerSpec = do withConnection handle $ \h -> do reply <- tcpTalk h (BSC.pack ("{\"method\":\"query\",\"sql\":\"" ++ replicate 40 'a' ++ "\"}")) asText (at "code" reply) `shouldBe` "too_large" + it "refuses oversized input before a newline and closes the connection" $ + withTcpServer defaultServerConfig{scMaxMessage = 32} $ \_ handle -> + withConnection handle $ \h -> do + BSC.hPutStr h (BSC.replicate 33 'a') + hFlush h + line <- timeout 1000000 (BSC.hGetLine h) + case line of + Nothing -> expectationFailure "oversized input waited for a newline" + Just raw -> asText (at "code" (fromMaybe A.Null (decode (BL.fromStrict raw)))) `shouldBe` "too_large" + tcpClosed h + it "accepts exact-limit CRLF frames and preserves pipelined requests" $ + withTcpServer defaultServerConfig{scMaxMessage = 17} $ \_ handle -> + withConnection handle $ \h -> do + BSC.hPutStr h "{\"method\":\"ping\"}\r\n{\"method\":\"ping\"}\n" + hFlush h + replicateM_ 2 $ do + raw <- timeout 1000000 (BSC.hGetLine h) + fmap (asText . at "status" . fromMaybe A.Null . decode . BL.fromStrict) raw `shouldBe` Just "pong" it "ping answers pong and quit closes the connection" $ withTcpServer defaultServerConfig $ \_ handle -> withConnection handle $ \h -> do @@ -1531,6 +1549,22 @@ transactionSpec = describe "server transactions" $ do rows <- mustSql session "select * from fresh" qrRowCount rows `shouldBe` 1 + it "does not start a transaction when its snapshot cannot be read" $ do + ref <- newMVar testDb + settings <- tempSettingsPath "snapshot-failure" + let base = memoryBackend testDatabaseName ref + -- | 保留测试后端并让快照读取失败 + failing name = (beWithDatabase base name) + { beSnapshot = pure (Left "data page unreadable") + , beWithDatabase = failing + } + session <- newSession (failing testDatabaseName) (T.pack testDatabaseName) settings + void (mustSql session "USE test") + result <- runStatementCoded session "BEGIN" + failureMessage result `shouldBe` "data page unreadable" + late <- runStatementCoded session "COMMIT" + failureMessage late `shouldBe` "no transaction in progress" + it "refuses nested BEGIN and COMMIT outside a transaction" $ withMemorySession $ \session -> do _ <- mustSql session "begin" nested <- runStatementCoded session "begin" diff --git a/chusql-web/static/api.js b/chusql-web/static/api.js index 1fdb9d4..bccc4e7 100644 --- a/chusql-web/static/api.js +++ b/chusql-web/static/api.js @@ -11,6 +11,12 @@ function isObject(value) { return value !== null && typeof value === 'object' && !Array.isArray(value); } +// 校验结果列名及每行的列数。 +function hasValidRows(body) { + return isObject(body) && Array.isArray(body.columns) && body.columns.every((column) => typeof column === 'string') + && Array.isArray(body.rows) && body.rows.every((row) => Array.isArray(row) && row.length === body.columns.length); +} + // 发一次请求并处理错误。 async function request(url, init, database) { const headers = new Headers(init && init.headers ? init.headers : undefined); @@ -114,21 +120,13 @@ export async function useDatabase(name, database) { // ---------------------------------------------------------------- 表结构 -// 取类型基名（去掉参数）。 -function columnKind(type) { - return String(type).split('(')[0].trim().toLowerCase(); -} - -const knownTypes = ['int', 'bigint', 'smallint', 'str', 'text', 'varchar', 'char', 'float', 'double', 'decimal', 'numeric', 'bool', 'date', 'timestamp', 'blob']; - // 校验并规整列描述。 function asColumns(value) { if (!Array.isArray(value)) throw new Error('服务器返回的列清单无法识别。'); return value.map((entry) => { - if (!isObject(entry) || typeof entry.name !== 'string' || typeof entry.type !== 'string') { + if (!isObject(entry) || typeof entry.name !== 'string' || typeof entry.type !== 'string' || !entry.type.trim()) { throw new Error('服务器返回的列描述无法识别。'); } - if (!knownTypes.includes(columnKind(entry.type))) throw new Error(`不支持的列类型：${entry.type}`); return { ...entry, name: entry.name, type: entry.type }; }); } @@ -197,13 +195,11 @@ export async function fetchRows({ database, table, limit, offset, sort, dir, fil params.set('filter', JSON.stringify(filters.map((f) => ({ column: f.column, value: f.expression })))); } const body = await request(`/api/tables/${encodeURIComponent(table)}/rows?${params}`, undefined, database); - if (!isObject(body) || !Array.isArray(body.columns) || !body.columns.every((c) => typeof c === 'string') - || !Array.isArray(body.rows) || typeof body.total !== 'number') { + if (!hasValidRows(body) || !Number.isSafeInteger(body.total) || body.total < 0) { throw new Error('服务器返回的分页数据无法识别。'); } const columns = body.columns; const rows = body.rows.map((raw) => { - if (!Array.isArray(raw)) throw new Error('服务器返回的行数据无法识别。'); return Object.fromEntries(columns.map((column, index) => [column, raw[index]])); }); return { columns, rows, total: body.total }; @@ -281,7 +277,7 @@ export async function query(sql, database, tabId) { const init = jsonInit('POST', { sql }); init.headers['X-ChuSQL-Console'] = console; const body = await request('/api/query', init, database); - if (!isObject(body) || !Array.isArray(body.columns) || !Array.isArray(body.rows)) { + if (!hasValidRows(body)) { throw new Error('服务器返回的查询结果无法识别。'); } const durationMs = performance.now() - started; diff --git a/chusql-web/test/runtime-values.mjs b/chusql-web/test/runtime-values.mjs index a83996f..9fb8087 100644 --- a/chusql-web/test/runtime-values.mjs +++ b/chusql-web/test/runtime-values.mjs @@ -1,7 +1,7 @@ import assert from 'node:assert/strict'; import { readFile } from 'node:fs/promises'; -// 检查前端复合值显示、SQL 生成和编码错误。 +// 检查复合值、SQL 生成、类型元数据和编码错误。 const source = await readFile(new URL('../static/core.js', import.meta.url), 'utf8'); const { cellText, sqlLiteral, typeLabel, editText } = await import(`data:text/javascript;base64,${Buffer.from(source).toString('base64')}`); @@ -17,4 +17,38 @@ assert.throws(() => sqlLiteral({ __chusql_runtime_v1: [1, [7, [typeInt]], [1, 2] assert.equal(typeLabel('runtime([1,[5,[[1,[]]]]])'), '[Int]'); assert.equal(editText({ __chusql_runtime_v1: [1, typeString, 'abc'] }), 'abc'); assert.throws(() => sqlLiteral({ __chusql_runtime_v1: [1, typeInt, 1.5] }), /integer/); -console.log('runtime values: 9 assertions passed'); +const apiSource = await readFile(new URL('../static/api.js', import.meta.url), 'utf8'); +const api = await import(`data:text/javascript;base64,${Buffer.from(apiSource).toString('base64')}`); +const columns = ['int', 'runtime([1,[5,[[1,[]]]]])', 'runtime([1,[6,[[4,[]]]]])', 'runtime([1,[7,[[1,[]],[4,[]]]]])', 'domain(age,int)'] + .map((type, index) => ({ name: `column_${index}`, type })); +const table = { table: 'typed', rowCount: 0, columns }; +const originalFetch = globalThis.fetch; +try { + globalThis.fetch = async (url) => new Response(JSON.stringify(url === '/api/tables' ? [table] : table)); + assert.deepEqual((await api.listTables('test'))[0].columns, columns); + assert.deepEqual((await api.getTable('test', 'typed')).columns, columns); + for (const type of [undefined, 42, ' ']) { + globalThis.fetch = async () => new Response(JSON.stringify([{ ...table, columns: [{ name: 'broken', type }] }])); + await assert.rejects(api.listTables('test'), /列描述/); + } + const page = { columns: ['id', 'value'], rows: [[1, null]], total: 1 }; + globalThis.fetch = async () => new Response(JSON.stringify(page)); + assert.deepEqual((await api.fetchRows({ database: 'test', table: 'typed', limit: 10, offset: 0 })).rows, + [{ id: 1, value: null }]); + assert.deepEqual((await api.query('SELECT id, value FROM typed', 'test')).rows, page.rows); + for (const invalid of [ + { ...page, rows: [[1]] }, { ...page, rows: [[1, null, 3]] }, + { ...page, rows: [null] }, { ...page, columns: ['id', 42] }, + ]) { + globalThis.fetch = async () => new Response(JSON.stringify(invalid)); + await assert.rejects(api.fetchRows({ database: 'test', table: 'typed', limit: 10, offset: 0 }), /分页数据/); + assert.equal((await api.query('SELECT id, value FROM typed', 'test')).type, 'error'); + } + for (const total of [-1, 1.5, '1']) { + globalThis.fetch = async () => new Response(JSON.stringify({ ...page, total })); + await assert.rejects(api.fetchRows({ database: 'test', table: 'typed', limit: 10, offset: 0 }), /分页数据/); + } +} finally { + globalThis.fetch = originalFetch; +} +console.log('runtime values and schemas: 27 assertions passed'); diff --git a/docs/architecture.md b/docs/architecture.md index 8619efa..ef3a504 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -117,7 +117,7 @@ TCP 连接在下一次请求时返回 unauthorized 并关闭；空闲连接不 ## 查询引擎扩展设计（尚未实现） -按 bound IR → 子查询物化 / 递归 CTE 推进；DOMAIN 约束见[类型文档](types.md#domain-约束设计尚未实现)。以下设计尚未实施，当前能力见前文；实施状态以 project-plan.txt 为准。 +按 bound IR → 子查询物化 / 递归 CTE 推进；DOMAIN 约束见[类型文档](types.md#domain-约束设计尚未实现)。以下设计尚未实施，当前能力见前文；实施状态以 agent-credientials/progress.txt 为准。 ### bound IR diff --git a/docs/normalization.md b/docs/normalization.md index 7afb07d..53844b2 100644 --- a/docs/normalization.md +++ b/docs/normalization.md @@ -92,7 +92,7 @@ schema 类型不证明业务意义上的原子性。不满足 1NF 时更高范 ## 后续入口设计（尚未实现） -以下 TCP、CLI 和 Web 接口尚未实现，拟议命令与端点目前不可用。实施状态以 project-plan.txt 为准。 +以下 TCP、CLI 和 Web 接口尚未实现，拟议命令与端点目前不可用。实施状态以 agent-credientials/progress.txt 为准。 ### 协议与鉴权 diff --git a/docs/release-signing.md b/docs/release-signing.md index c669820..77d2ddb 100644 --- a/docs/release-signing.md +++ b/docs/release-signing.md @@ -1,6 +1,6 @@ # 发布签名与验证设计 -本文定义尚未实现的发行安全方案。当前操作见[安装文档](install.md)，实施状态以 project-plan.txt 为准。先实现离线验证，再接入在线安装；该工作可独立于查询和存储扩展推进。 +本文定义尚未实现的发行安全方案。当前操作见[安装文档](install.md)，实施状态以 agent-credientials/progress.txt 为准。先实现离线验证，再接入在线安装；该工作可独立于查询和存储扩展推进。 ## 选择与信任入口 diff --git a/gate-check/README.md b/gate-check/README.md index 506072e..9db6a79 100644 --- a/gate-check/README.md +++ b/gate-check/README.md @@ -42,5 +42,4 @@ Web 步骤会先显式刷新该工程里的 model、engine、interface 与 serve ## 约定 - 门禁由项目所有者手动运行；agent 只跑模块级测试，不跑全量门禁。 -- 期望基线（随里程碑更新）：Engine 373 / Server 102 / Web 56 / CLI 13 / Bootstrap 9 / Rust 154。 -- 计数与主线进度以 `agent-credientials/project-plan.txt` 为准。 +- 计数、验收状态与主线进度以 `agent-credientials/progress.txt` 为准。