packages feed

valiant 0.1.0.0 → 0.1.0.1

raw patch · 8 files changed

+242/−83 lines, 8 filesdep −psqueuesdep ~pg-wire

Dependencies removed: psqueues

Dependency ranges changed: pg-wire

Files

CHANGELOG.md view
@@ -3,6 +3,23 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.1.0.1] - 2026-04-29++### Changed++- Switched the per-connection prepared-statement cache used by+  `Valiant.Execute` and `Valiant.Batch` from a HashPSQ-based LRU to+  SIEVE ([NSDI'24](https://www.usenix.org/conference/nsdi24/presentation/zhang-yazhuo)).+  In standalone microbenchmarks, SIEVE is 2.3-3.8x faster than the+  previous LRU across fill, hit-only, mixed, eviction-storm,+  uniform-random, and Zipf-skewed workloads. No public API change.+- Updated `pg-wire` lower bound to `==0.2.*` (required to pick up the+  SIEVE-backed `Connection` and the `PgWire.Cache.Sieve` exports).++### Removed++- `psqueues` dependency.+ ## [0.1.0.0] - 2026-04-25  Initial release. Runtime library on top of@@ -28,4 +45,5 @@   indirection for `Int32` / `Int64` / `Double` / `Bool` and tuples of   those. +[0.1.0.1]: https://github.com/joshburgess/valiant/releases/tag/valiant-v0.1.0.1 [0.1.0.0]: https://github.com/joshburgess/valiant/releases/tag/v0.1.0.0
src/Valiant.hs view
@@ -170,6 +170,9 @@   , withTransactionRetry   , withTransactionRetryIf   , withSavepoint+  , setLocal+  , setLocals+  , withTransactionContext      -- * Valiant monad (optional convenience)     -- | For the @ReaderT Pool IO@ monad with lifted @*M@ functions,@@ -288,6 +291,7 @@   , isForeignKeyViolation   , isSerializationError   , isDeadlockError+  , isFatal      -- * Binary codec type classes     -- | Low-level encode\/decode classes for individual PostgreSQL types.@@ -365,7 +369,7 @@ -- "Dynamic SQL" haddock note above). The bare import ensures haddock can -- resolve the "Valiant.Dynamic" cross-reference link. import Valiant.Dynamic ()-import Valiant.Error (ConstraintViolation (..), catchConstraintViolation, constraintViolation, isDeadlockError, isForeignKeyViolation, isSerializationError, isUniqueViolation, pgErrorOf, sqlState)+import Valiant.Error (ConstraintViolation (..), catchConstraintViolation, constraintViolation, isDeadlockError, isFatal, isForeignKeyViolation, isSerializationError, isUniqueViolation, pgErrorOf, sqlState) import Valiant.Execute (execute, executeBatch, executeMany, executeReturning, executeReturningMany, fetchAll, fetchAllFast, fetchAllUnboxed, fetchAllVec, fetchAllWith, fetchBatchAll, fetchBatchOne, fetchExists, fetchFirst, fetchOne, fetchOneFast, fetchOneOr, fetchOneOrThrow, fetchScalar, forEach, rawExecute, rawFetchAll, rawFetchOne) import Valiant.FromRowFast (FromRowFast (..), DecodeColumnFast (..)) import Valiant.Fold (RowFold (..), executeWithFold)@@ -380,7 +384,7 @@ import Valiant.Streaming (CursorState, fetchAllCursor, fetchBatch, withCursor) import Valiant.ToParams (EncodeField (..), ToParams (..)) import PgWire.Binary.Types (PgDecode (..), PgEncode (..))-import Valiant.Transaction (IsolationLevel (..), Transaction (..), TransactionMode (..), defaultTransactionMode, withDeferrableTransaction, withReadOnlyTransaction, withSavepoint, withTransaction, withTransactionConn, withTransactionLevel, withTransactionLevelConn, withTransactionMode, withTransactionModeConn, withTransactionRetry, withTransactionRetryIf, withTransaction_)+import Valiant.Transaction (IsolationLevel (..), Transaction (..), TransactionMode (..), defaultTransactionMode, setLocal, setLocals, withDeferrableTransaction, withReadOnlyTransaction, withSavepoint, withTransaction, withTransactionConn, withTransactionContext, withTransactionLevel, withTransactionLevelConn, withTransactionMode, withTransactionModeConn, withTransactionRetry, withTransactionRetryIf, withTransaction_) import PgWire.Cancel (cancelQuery, withQueryTimeout) import PgWire.Connection (Connection, close, connect, connectString, simpleQuery, withConnection) import PgWire.Connection.Config (ConnConfig (..), TlsMode (..), defaultConnConfig)
src/Valiant/Batch.hs view
@@ -25,11 +25,11 @@ import Data.ByteString (ByteString) import Data.ByteString.Char8 qualified as BS8 import Data.IORef-import Data.HashPSQ qualified as PSQ import Data.Vector (Vector) import Data.Vector qualified as V-import Data.Word (Word32, Word64)+import Data.Word (Word32) import PgWire.Async (Request (..), Response (..), ResponseCollector (..), submitRequest)+import PgWire.Cache.Sieve qualified as Sieve import PgWire.Binary.Types (PgEncode (..)) import PgWire.Connection (Connection (..)) import PgWire.Error (PgWireError (..), throwPgWire)@@ -102,39 +102,24 @@  -- Internal helpers -------------------------------------------------------- --- | Bump the monotonic tick counter and return the new value.-nextTick :: Connection -> IO Word64-nextTick conn = atomicModifyIORef' (connStmtTick conn) (\n -> (n + 1, n + 1))- ensurePreparedRaw :: Connection -> ByteString -> Vector Word32 -> IO ByteString ensurePreparedRaw conn sql oids = do-  cache <- readIORef (connStmtCache conn)-  case PSQ.lookup sql cache of-    Just (_prio, name) -> do-      tick <- nextTick conn-      modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)-      pure name+  mname <- Sieve.lookup (connStmtCache conn) sql+  case mname of+    Just name -> pure name     Nothing -> do-      evictIfNeeded conn cache       counter <- atomicModifyIORef' (connStmtCounter conn) (\n -> (n + 1, n))       let name = "s" <> BS8.pack (show counter)       resp <- submitRequest (connAsync conn) $ ReqPrepare (Parse name sql oids)       case resp of         RespParsed -> do-          tick <- nextTick conn-          modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)+          evicted <- Sieve.insert (connStmtCache conn) sql name+          case evicted of+            Nothing -> pure ()+            Just (_oldSql, oldName) -> do+              cresp <- submitRequest (connAsync conn) $ ReqClose (Close DescribeStatement oldName)+              case cresp of+                RespClosed -> pure ()+                _ -> pure ()  -- best effort           pure name         _ -> throwPgWire (ProtocolError "ensurePreparedRaw: unexpected response type")---- | Evict the least recently used statement if the cache is full.-evictIfNeeded :: Connection -> PSQ.HashPSQ ByteString Word64 ByteString -> IO ()-evictIfNeeded conn cache-  | PSQ.size cache < connMaxPreparedStatements conn = pure ()-  | otherwise = case PSQ.findMin cache of-      Nothing -> pure ()-      Just (oldSql, _prio, oldName) -> do-        resp <- submitRequest (connAsync conn) $ ReqClose (Close DescribeStatement oldName)-        case resp of-          RespClosed -> pure ()-          _ -> pure () -- best effort-        modifyIORef' (connStmtCache conn) (PSQ.delete oldSql)
src/Valiant/Error.hs view
@@ -42,12 +42,15 @@   , isUndefinedTable   , isUndefinedColumn   , isSyntaxError++    -- * Connection fatality+  , isFatal   ) where  import Control.Exception (catch, throwIO) import Data.Maybe (fromMaybe) import Data.ByteString (ByteString)-import PgWire.Error (PgWireError (..))+import PgWire.Error (PgWireError (..), isFatal) import PgWire.Protocol.Backend (PgError (..))  -- | Extract the SQLSTATE code from a 'PgWireError', if it wraps a
src/Valiant/Execute.hs view
@@ -55,14 +55,13 @@ import Data.ByteString.Char8 qualified as BS8 import Data.IORef import Data.Int (Int64)-import Data.HashPSQ (HashPSQ)-import Data.HashPSQ qualified as PSQ import Data.Maybe (fromMaybe) import Data.Vector (Vector) import Data.Vector qualified as V import Data.Vector.Unboxed qualified as U-import Data.Word (Word32, Word64)+import Data.Word (Word32) import PgWire.Async (Request (..), Response (..), ResponseCollector (..), submitRequest, submitExclusive)+import PgWire.Cache.Sieve qualified as Sieve import PgWire.Connection (Connection (..)) import PgWire.Protocol.Oid qualified as Oid import PgWire.Error (PgWireError (..), throwPgWire)@@ -502,14 +501,10 @@ -- | Look up or allocate a statement name for raw SQL (same cache as typed). lookupOrAllocRaw :: Connection -> ByteString -> IO (ByteString, Bool) lookupOrAllocRaw conn sql = do-  cache <- readIORef (connStmtCache conn)-  case PSQ.lookup sql cache of-    Just (_prio, name) -> do-      tick <- nextTick conn-      modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)-      pure (name, False)+  mname <- Sieve.lookup (connStmtCache conn) sql+  case mname of+    Just name -> pure (name, False)     Nothing -> do-      evictIfNeeded conn cache       counter <- atomicModifyIORef' (connStmtCounter conn) (\n -> (n + 1, n))       let name = "s" <> BS8.pack (show counter)       pure (name, True)@@ -662,7 +657,9 @@       pure rows     _ -> throwPgWire (ProtocolError "fetchRows: unexpected response type") --- | Check the statement cache. On miss, allocate a name and evict if needed.+-- | Check the statement cache. On miss, allocate a fresh name; the actual+-- cache insert (and any eviction it triggers) happens in 'cacheStmt' once+-- Parse has succeeded server-side. -- When prepared statements are disabled (PgBouncer compatibility), always -- returns the unnamed statement ("", True) so queries are parsed each time. lookupOrAllocStmt :: Connection -> Statement p r -> IO (ByteString, Bool)@@ -671,27 +668,31 @@       -- Unprepared mode: always use unnamed statement, always parse       pure ("", True)   | otherwise = do-      cache <- readIORef (connStmtCache conn)       let sql = stmtSQL stmt-      case PSQ.lookup sql cache of-        Just (_prio, name) -> do-          tick <- nextTick conn-          modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)-          pure (name, False)+      mname <- Sieve.lookup (connStmtCache conn) sql+      case mname of+        Just name -> pure (name, False)         Nothing -> do-          evictIfNeeded conn cache           counter <- atomicModifyIORef' (connStmtCounter conn) (\n -> (n + 1, n))           let name = "s" <> BS8.pack (show counter)           pure (name, True) --- | Cache a statement name after successful Parse.+-- | Cache a statement name after successful Parse, returning any entry+-- that the SIEVE eviction policy chose to discard so the caller can release+-- the corresponding server-side prepared statement. -- No-op when prepared statements are disabled (unnamed statements are never cached). cacheStmt :: Connection -> ByteString -> ByteString -> IO () cacheStmt conn sql name   | not (connPreparedStatements conn) = pure ()   | otherwise = do-      tick <- nextTick conn-      modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)+      evicted <- Sieve.insert (connStmtCache conn) sql name+      case evicted of+        Nothing -> pure ()+        Just (_oldSql, oldName) -> do+          resp <- submitRequest (connAsync conn) $ ReqClose (Close DescribeStatement oldName)+          case resp of+            RespClosed -> pure ()+            _ -> pure () -- best effort  -- | Decode a list of raw row vectors into typed values. decodeRows :: (Vector (Maybe ByteString) -> Either String r) -> [Vector (Maybe ByteString)] -> IO [r]@@ -709,43 +710,20 @@ -- callers that need the statement name before building messages (Pipeline, Fold). ensurePrepared :: Connection -> Statement p r -> IO ByteString ensurePrepared conn stmt = do-  cache <- readIORef (connStmtCache conn)   let sql = stmtSQL stmt-  case PSQ.lookup sql cache of-    Just (_prio, name) -> do-      tick <- nextTick conn-      modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)-      pure name+  mname <- Sieve.lookup (connStmtCache conn) sql+  case mname of+    Just name -> pure name     Nothing -> do-      evictIfNeeded conn cache       counter <- atomicModifyIORef' (connStmtCounter conn) (\n -> (n + 1, n))       let name = "s" <> BS8.pack (show counter)           oids = V.map Oid.unOid (stmtParamOids stmt)       resp <- submitRequest (connAsync conn) $ ReqPrepare (Parse name sql oids)       case resp of         RespParsed -> do-          tick <- nextTick conn-          modifyIORef' (connStmtCache conn) (PSQ.insert sql tick name)+          cacheStmt conn sql name           pure name         _ -> throwPgWire (ProtocolError "ensurePrepared: unexpected response type")---- | Bump the monotonic tick counter and return the new value.-nextTick :: Connection -> IO Word64-nextTick conn = atomicModifyIORef' (connStmtTick conn) (\n -> (n + 1, n + 1))---- | If the cache has reached its limit, evict the least recently used--- prepared statement (the entry with the lowest tick priority).-evictIfNeeded :: Connection -> HashPSQ ByteString Word64 ByteString -> IO ()-evictIfNeeded conn cache-  | PSQ.size cache < connMaxPreparedStatements conn = pure ()-  | otherwise = case PSQ.findMin cache of-      Nothing -> pure ()-      Just (oldSql, _prio, oldName) -> do-        resp <- submitRequest (connAsync conn) $ ReqClose (Close DescribeStatement oldName)-        case resp of-          RespClosed -> pure ()-          _ -> pure () -- best effort-        modifyIORef' (connStmtCache conn) (PSQ.delete oldSql)  ------------------------------------------------------------------------ -- Streaming batch helpers (exclusive mode, direct wire access)
src/Valiant/Transaction.hs view
@@ -25,6 +25,15 @@   , withTransactionRetry   , withTransactionRetryIf   , withSavepoint++    -- * Session variables (SET LOCAL)+    -- | Set transaction-scoped configuration parameters using @SET LOCAL@.+    -- Values are automatically reset when the transaction ends (commit or+    -- rollback). This is the standard mechanism for passing context to+    -- Row Level Security (RLS) policies in multi-tenant applications.+  , setLocal+  , setLocals+  , withTransactionContext   ) where  import Control.Exception (SomeException, catch, mask, onException, throwIO, try)@@ -33,7 +42,7 @@ import Data.ByteString.Char8 qualified as BS8 import Data.IORef import Data.Word (Word64)-import PgWire.Connection (Connection, simpleQuery)+import PgWire.Connection (Connection, escapeLiteral, simpleQuery) import PgWire.Error (PgWireError (..)) import PgWire.Protocol.Backend (PgError (..)) import PgWire.Pool (Pool, withResource)@@ -310,3 +319,72 @@ freshSavepointName = do   n <- atomicModifyIORef' savepointCounter (\n -> (n + 1, n))   pure ("valiant_sp_" <> BS8.pack (show n))++------------------------------------------------------------------------+-- SET LOCAL helpers+------------------------------------------------------------------------++-- | Set a transaction-scoped configuration parameter using @SET LOCAL@.+--+-- The value is automatically reset when the transaction ends. The value is+-- safely escaped to prevent SQL injection.+--+-- This is the standard way to pass context to PostgreSQL Row Level Security+-- (RLS) policies, e.g. setting the current tenant or user claims.+--+-- @+-- withTransaction pool $ \\tx -> do+--   setLocal tx \"app.current_tenant\" \"acme-corp\"+--   setLocal tx \"request.jwt.claim.sub\" \"user-123\"+--   fetchAll (txConn tx) tenantQuery ()+-- @+--+-- Must be called within a transaction — @SET LOCAL@ outside a transaction+-- has no effect (the value is discarded immediately).+setLocal :: Transaction -> ByteString -> ByteString -> IO ()+setLocal tx name value = do+  let conn = txConn tx+      escaped = escapeLiteral conn value+      stmt = "SET LOCAL " <> name <> " = " <> escaped+  void $ simpleQuery conn stmt++-- | Set multiple transaction-scoped parameters at once.+--+-- @+-- withTransaction pool $ \\tx -> do+--   setLocals tx+--     [ (\"app.current_tenant\", tenantId)+--     , (\"app.current_role\", role)+--     , (\"request.jwt.claim.sub\", userId)+--     ]+--   fetchAll (txConn tx) rlsQuery ()+-- @+setLocals :: Transaction -> [(ByteString, ByteString)] -> IO ()+setLocals tx = mapM_ (uncurry (setLocal tx))++-- | Run a transaction with pre-set context variables.+--+-- Combines 'withTransaction' and 'setLocals' into a single call. The context+-- variables are set immediately after @BEGIN@ and before the action runs.+--+-- This is the recommended pattern for multi-tenant applications using+-- Row Level Security:+--+-- @+-- withTransactionContext pool+--   [ (\"app.current_tenant\", tenantId)+--   , (\"request.jwt.claim.sub\", userId)+--   ]+--   $ \\tx -> do+--     users <- fetchAll (txConn tx) listTenantUsers ()+--     ...+-- @+withTransactionContext+  :: Pool+  -> [(ByteString, ByteString)]+  -> (Transaction -> IO a)+  -> IO a+withTransactionContext pool ctx action =+  withTransaction pool $ \tx -> do+    setLocals tx ctx+    action tx
test/Valiant/ErrorSpec.hs view
@@ -141,3 +141,97 @@             _ -> pure "wrong type")           (simpleQuery conn "INSERT" >> pure "no error")         result `shouldBe` ("caught" :: String)++  describe "isFatal" $ do+    it "ConnectionError is fatal" $+      isFatal (ConnectionError "test") `shouldBe` True++    it "AuthError is fatal" $+      isFatal (AuthError "bad password") `shouldBe` True++    it "ProtocolError is fatal" $+      isFatal (ProtocolError "unexpected message") `shouldBe` True++    it "ConnectionDead is fatal" $+      isFatal ConnectionDead `shouldBe` True++    it "PoolTimeout is not fatal" $+      isFatal PoolTimeout `shouldBe` False++    it "PoolClosed is not fatal" $+      isFatal PoolClosed `shouldBe` False++    it "DecodeError is not fatal" $+      isFatal (DecodeError "bad row") `shouldBe` False++    it "constraint violation (23505) is not fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "23505" "duplicate key"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` False++    it "syntax error (42601) is not fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "42601" "syntax error"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` False++    it "serialization failure (40001) is not fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "40001" "could not serialize"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` False++    it "connection exception (08006) is fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "08006" "connection failure"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` True++    it "admin shutdown (57P01) is fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "57P01" "admin shutdown"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` True++    it "crash recovery (57P02) is fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "57P02" "crash shutdown"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` True++    it "system IO error (58030) is fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "58030" "io_error"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` True++    it "internal error (XX000) is fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "XX000" "internal error"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` True++    it "undefined table (42P01) is not fatal" $ do+      let cfg = defaultMockConfig+            { mockQueryHandler = errorHandler "42P01" "relation does not exist"+            }+      withMockConn cfg $ \conn -> do+        err <- triggerError conn+        isFatal err `shouldBe` False
valiant.cabal view
@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            valiant-version:         0.1.0.0+version:         0.1.0.1 synopsis:        Compile-time checked SQL for Haskell, runtime library description:   Runtime library for valiant. Provides the @Statement@ type, binary format@@ -84,7 +84,7 @@    build-depends:     -- Wire protocol driver-    , pg-wire               ==0.1.*+    , pg-wire               ==0.2.*      -- Core     , base                  >=4.17    && <5@@ -92,7 +92,6 @@     , text                  >=2.0     && <2.2     , containers            >=0.6     && <0.8     , hashable              >=1.4     && <1.6-    , psqueues              >=0.2.8   && <0.3     , vector                >=0.13    && <0.14     , time                  >=1.12    && <1.15     , deepseq               >=1.4     && <1.6