diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,11 @@ -# 1.2 +# v1.2.2.1 +- Conform to the new `hasql` API (v2.0) + +# v1.2 + - Removed the `unpreparedTransaction` session because the same effects can now be achieved via the connection settings in Hasql -# 1.1 +# v1.1 - Add automatic retry on deadlock errors (code 40P01) diff --git a/hasql-transaction.cabal b/hasql-transaction.cabal --- a/hasql-transaction.cabal +++ b/hasql-transaction.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-transaction -version: 1.2.2 +version: 1.2.2.1 category: Hasql, Database, PostgreSQL synopsis: Composable abstraction over retryable transactions for Hasql @@ -85,8 +85,8 @@ other-modules: Hasql.Transaction.Config Hasql.Transaction.Private.Prelude - Hasql.Transaction.Private.SQL Hasql.Transaction.Private.Sessions + Hasql.Transaction.Private.SQL Hasql.Transaction.Private.Statements Hasql.Transaction.Private.Transaction @@ -95,27 +95,35 @@ bytestring >=0.10 && <0.13, bytestring-tree-builder >=0.2.7.8 && <0.3, contravariant >=1.3 && <2, - hasql >=1.10 && <1.11, + hasql >=1.10 && <1.11 || >=2.0 && <2.1, mtl >=2.2 && <3, text >=1 && <3, transformers >=0.5 && <0.7, -test-suite conflicts-test +test-suite test import: test type: exitcode-stdio-1.0 - hs-source-dirs: src/conflicts-test + hs-source-dirs: src/integration-tests main-is: Main.hs other-modules: - Main.Statements - Main.Transactions + Helpers.Adapters + Helpers.Hooks + Helpers.Scripts + Helpers.Statements + Helpers.Transactions + Specs.ConflictsSpec + Specs.SpecHook - ghc-options: - -O2 - -threaded - -with-rtsopts=-N + build-tool-depends: + hspec-discover:hspec-discover ^>=2.11.12 build-depends: async >=2.1 && <3, - hasql >=1.9, + hasql, hasql-transaction, + hspec >=2.6 && <3, + pqi ^>=1.0, + pqi-ffi ^>=1.0, + pqi-native ^>=1.0, rerebase >=1.11 && <2, + testcontainers-postgresql >=0.2 && <0.3, diff --git a/src/conflicts-test/Main.hs b/src/conflicts-test/Main.hs deleted file mode 100644 --- a/src/conflicts-test/Main.hs +++ /dev/null @@ -1,121 +0,0 @@ -module Main where - -import Control.Concurrent.Async qualified as F -import Hasql.Connection qualified as A -import Hasql.Connection.Settings qualified as H -import Hasql.Session qualified as B -import Hasql.Transaction qualified as C -import Hasql.Transaction.Sessions qualified as G -import Main.Statements qualified as D -import Main.Transactions qualified as E -import Prelude - -main :: IO () -main = - bracket acquire release use - where - acquire = - (,) <$> acquire <*> acquire - where - acquire = - join - $ fmap (either (fail . show) return) - $ A.acquire connectionSettings - where - connectionSettings = - H.hostAndPort "localhost" 5432 - <> H.user "postgres" - <> H.password "postgres" - <> H.dbname "postgres" - release (connection1, connection2) = - do - transaction connection1 E.dropSchema - A.release connection1 - A.release connection2 - use (connection1, connection2) = - do - try (transaction connection1 E.dropSchema) :: IO (Either SomeException ()) - transaction connection1 E.createSchema - success <- fmap and (traverse runTest tests) - if success - then exitSuccess - else exitFailure - where - runTest test = - test connection1 connection2 - tests = - [readAndWriteTransactionsTest, transactionsTest, transactionsNoRetryTest, transactionAndQueryTest] - -session :: A.Connection -> B.Session a -> IO a -session connection session = - A.use connection session - >>= either (fail . show) return - -transaction :: A.Connection -> C.Transaction a -> IO a -transaction connection transaction = - session connection (G.transaction G.RepeatableRead G.Write transaction) - -transactionNoRetry :: A.Connection -> C.Transaction a -> IO a -transactionNoRetry connection transaction = - session connection (G.transactionNoRetry G.RepeatableRead G.Write transaction) - -type Test = - A.Connection -> A.Connection -> IO Bool - -transactionsTest :: Test -transactionsTest connection1 connection2 = - do - id1 <- session connection1 (B.statement 0 D.createAccount) - id2 <- session connection1 (B.statement 0 D.createAccount) - async1 <- F.async (replicateM_ 1000 (transaction connection1 (E.transfer id1 id2 1))) - async2 <- F.async (replicateM_ 1000 (transaction connection2 (E.transfer id1 id2 1))) - F.wait async1 - F.wait async2 - balance1 <- session connection1 (B.statement id1 D.getBalance) - balance2 <- session connection1 (B.statement id2 D.getBalance) - traceShowM balance1 - traceShowM balance2 - return (balance1 == Just 2000 && balance2 == Just (-2000)) - -transactionsNoRetryTest :: Test -transactionsNoRetryTest connection1 connection2 = - do - id1 <- session connection1 (B.statement 0 D.createAccount) - id2 <- session connection1 (B.statement 0 D.createAccount) - async1 <- F.async (replicateM_ 1000 (transactionNoRetry connection1 (E.transfer id1 id2 1))) - async2 <- F.async (replicateM_ 1000 (transactionNoRetry connection2 (E.transfer id1 id2 1))) - result1 <- F.waitCatch async1 - result2 <- F.waitCatch async2 - let serialError = sequenceA [result1, result2] - traceShowM serialError - return $ either (("40001" `isInfixOf`) . show) (pure False) serialError - -readAndWriteTransactionsTest :: Test -readAndWriteTransactionsTest connection1 connection2 = - do - id1 <- session connection1 (B.statement 0 D.createAccount) - id2 <- session connection1 (B.statement 0 D.createAccount) - async1 <- F.async (replicateM_ 1000 (transaction connection1 (E.transfer id1 id2 1))) - async2 <- F.async (replicateM_ 1000 (transaction connection2 (C.statement id1 D.getBalance))) - F.wait async1 - F.wait async2 - balance1 <- session connection1 (B.statement id1 D.getBalance) - balance2 <- session connection1 (B.statement id2 D.getBalance) - traceShowM balance1 - traceShowM balance2 - return (balance1 == Just 1000 && balance2 == Just (-1000)) - -transactionAndQueryTest :: Test -transactionAndQueryTest connection1 connection2 = - do - id1 <- session connection1 (B.statement 0 D.createAccount) - id2 <- session connection1 (B.statement 0 D.createAccount) - async1 <- F.async (transaction connection1 (E.transferTimes 200 id1 id2 1)) - async2 <- F.async (session connection2 (replicateM_ 200 (B.statement (id1, 1) D.modifyBalance))) - F.wait async1 - F.wait async2 - balance1 <- session connection1 (B.statement id1 D.getBalance) - balance2 <- session connection1 (B.statement id2 D.getBalance) - traceShowM balance1 - traceShowM balance2 - return (balance1 == Just 400 && balance2 == Just (-200)) diff --git a/src/conflicts-test/Main/Statements.hs b/src/conflicts-test/Main/Statements.hs deleted file mode 100644 --- a/src/conflicts-test/Main/Statements.hs +++ /dev/null @@ -1,41 +0,0 @@ -module Main.Statements where - -import Hasql.Decoders qualified as D -import Hasql.Encoders qualified as E -import Hasql.Statement -import Prelude - -createAccountTable :: Statement () () -createAccountTable = - unpreparable sql E.noParams D.noResult - where - sql = - "create table account (id bigserial not null, balance numeric not null, primary key (id))" - -dropAccountTable :: Statement () () -dropAccountTable = - unpreparable - "drop table account" - E.noParams - D.noResult - -createAccount :: Statement Scientific Int64 -createAccount = - preparable - "insert into account (balance) values ($1) returning id" - ((E.param . E.nonNullable) E.numeric) - (D.singleRow ((D.column . D.nonNullable) D.int8)) - -modifyBalance :: Statement (Int64, Scientific) Bool -modifyBalance = - preparable - "update account set balance = balance + $2 where id = $1" - ((fst >$< (E.param . E.nonNullable) E.int8) <> (snd >$< (E.param . E.nonNullable) E.numeric)) - (fmap (> 0) D.rowsAffected) - -getBalance :: Statement Int64 (Maybe Scientific) -getBalance = - preparable - "select balance from account where id = $1" - ((E.param . E.nonNullable) E.int8) - (D.rowMaybe ((D.column . D.nonNullable) D.numeric)) diff --git a/src/conflicts-test/Main/Transactions.hs b/src/conflicts-test/Main/Transactions.hs deleted file mode 100644 --- a/src/conflicts-test/Main/Transactions.hs +++ /dev/null @@ -1,27 +0,0 @@ -module Main.Transactions where - -import Hasql.Transaction -import Main.Statements qualified as A -import Prelude - -createSchema :: Transaction () -createSchema = - do - statement () A.createAccountTable - -dropSchema :: Transaction () -dropSchema = - do - statement () A.dropAccountTable - -transfer :: Int64 -> Int64 -> Scientific -> Transaction Bool -transfer id1 id2 amount = - do - success <- statement (id1, amount) A.modifyBalance - if success - then statement (id2, negate amount) A.modifyBalance - else return False - -transferTimes :: Int -> Int64 -> Int64 -> Scientific -> Transaction () -transferTimes times id1 id2 amount = - replicateM_ times (transfer id1 id2 amount) diff --git a/src/integration-tests/Helpers/Adapters.hs b/src/integration-tests/Helpers/Adapters.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Adapters.hs @@ -0,0 +1,30 @@ +module Helpers.Adapters + ( adapters, + byAdapter, + hook, + ) +where + +import Pqi qualified +import Pqi.Ffi qualified +import Pqi.Native qualified +import Prelude +import Test.Hspec + +adapters :: [Pqi.Adapter] +adapters = + [ Pqi.Ffi.adapter, + Pqi.Native.adapter + ] + +-- | Run the given spec-building function once per available Pqi adapter, +-- nesting each run under a @describe@ named after the adapter. +byAdapter :: (Pqi.Adapter -> Spec) -> Spec +byAdapter f = + for_ adapters \adapter -> + describe (toList (Pqi.name adapter)) (f adapter) + +hook :: SpecWith Pqi.Adapter -> Spec +hook hookedSpec = + byAdapter \adapter -> + mapSubject (const adapter) hookedSpec diff --git a/src/integration-tests/Helpers/Hooks.hs b/src/integration-tests/Helpers/Hooks.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Hooks.hs @@ -0,0 +1,19 @@ +-- | Hooks for Hspec. +module Helpers.Hooks where + +import Data.Bool +import Prelude hiding (Handler) +import TestcontainersPostgresql qualified + +-- | Testing action in the scope of the host name and port of a running fresh isolated postgres server. +type Handler = (Text, Word16) -> IO () + +postgres17 :: Handler -> IO () +postgres17 handler = + TestcontainersPostgresql.run + TestcontainersPostgresql.Config + { forwardLogs = False, + tagName = "postgres:17", + auth = TestcontainersPostgresql.TrustAuth + } + (\(host, portInt) -> handler (host, fromIntegral portInt)) diff --git a/src/integration-tests/Helpers/Scripts.hs b/src/integration-tests/Helpers/Scripts.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Scripts.hs @@ -0,0 +1,61 @@ +module Helpers.Scripts + ( ScopeParams, + onConnectionPair, + session, + transaction, + transactionNoRetry, + ) +where + +import Hasql.Connection qualified as Connection +import Hasql.Connection.Settings qualified as Settings +import Hasql.Session qualified as Session +import Hasql.Transaction qualified as Transaction +import Hasql.Transaction.Sessions qualified as Transaction.Sessions +import Helpers.Transactions qualified as Transactions +import Pqi qualified +import Prelude + +-- | +-- Adapter, host and port of a running isolated postgres server. +type ScopeParams = (Pqi.Adapter, Text, Word16) + +-- | +-- Acquire a pair of connections against a fresh copy of the schema, +-- releasing them and dropping the schema once the action completes. +onConnectionPair :: ScopeParams -> (Connection.Connection -> Connection.Connection -> IO ()) -> IO () +onConnectionPair (adapter, host, port) action = + bracket acquire release use + where + acquire = + (,) <$> acquireConnection <*> acquireConnection + where + acquireConnection = + Connection.acquire adapter connectionSettings + >>= either (fail . show) return + connectionSettings = + Settings.hostAndPort host port + <> Settings.user "postgres" + <> Settings.password "postgres" + <> Settings.dbname "postgres" + release (connection1, connection2) = do + transaction connection1 Transactions.dropSchema + Connection.release connection1 + Connection.release connection2 + use (connection1, connection2) = do + _ <- try (transaction connection1 Transactions.dropSchema) :: IO (Either SomeException ()) + transaction connection1 Transactions.createSchema + action connection1 connection2 + +session :: Connection.Connection -> Session.Session a -> IO a +session connection theSession = + Connection.use connection theSession + >>= either (fail . show) return + +transaction :: Connection.Connection -> Transaction.Transaction a -> IO a +transaction connection theTransaction = + session connection (Transaction.Sessions.transaction Transaction.Sessions.RepeatableRead Transaction.Sessions.Write theTransaction) + +transactionNoRetry :: Connection.Connection -> Transaction.Transaction a -> IO a +transactionNoRetry connection theTransaction = + session connection (Transaction.Sessions.transactionNoRetry Transaction.Sessions.RepeatableRead Transaction.Sessions.Write theTransaction) diff --git a/src/integration-tests/Helpers/Statements.hs b/src/integration-tests/Helpers/Statements.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Statements.hs @@ -0,0 +1,41 @@ +module Helpers.Statements where + +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Statement +import Prelude + +createAccountTable :: Statement () () +createAccountTable = + unpreparable sql E.noParams D.noResult + where + sql = + "create table account (id bigserial not null, balance numeric not null, primary key (id))" + +dropAccountTable :: Statement () () +dropAccountTable = + unpreparable + "drop table account" + E.noParams + D.noResult + +createAccount :: Statement Scientific Int64 +createAccount = + preparable + "insert into account (balance) values ($1) returning id" + ((E.param . E.nonNullable) E.numeric) + (D.singleRow ((D.column . D.nonNullable) D.int8)) + +modifyBalance :: Statement (Int64, Scientific) Bool +modifyBalance = + preparable + "update account set balance = balance + $2 where id = $1" + ((fst >$< (E.param . E.nonNullable) E.int8) <> (snd >$< (E.param . E.nonNullable) E.numeric)) + (fmap (> 0) D.rowsAffected) + +getBalance :: Statement Int64 (Maybe Scientific) +getBalance = + preparable + "select balance from account where id = $1" + ((E.param . E.nonNullable) E.int8) + (D.rowMaybe ((D.column . D.nonNullable) D.numeric)) diff --git a/src/integration-tests/Helpers/Transactions.hs b/src/integration-tests/Helpers/Transactions.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Helpers/Transactions.hs @@ -0,0 +1,25 @@ +module Helpers.Transactions where + +import Hasql.Transaction +import Helpers.Statements qualified as Statements +import Prelude + +createSchema :: Transaction () +createSchema = + statement () Statements.createAccountTable + +dropSchema :: Transaction () +dropSchema = + statement () Statements.dropAccountTable + +transfer :: Int64 -> Int64 -> Scientific -> Transaction Bool +transfer id1 id2 amount = + do + success <- statement (id1, amount) Statements.modifyBalance + if success + then statement (id2, negate amount) Statements.modifyBalance + else return False + +transferTimes :: Int -> Int64 -> Int64 -> Scientific -> Transaction () +transferTimes times id1 id2 amount = + replicateM_ times (transfer id1 id2 amount) diff --git a/src/integration-tests/Main.hs b/src/integration-tests/Main.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Main.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-} diff --git a/src/integration-tests/Specs/ConflictsSpec.hs b/src/integration-tests/Specs/ConflictsSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/ConflictsSpec.hs @@ -0,0 +1,62 @@ +module Specs.ConflictsSpec where + +import Control.Concurrent.Async qualified as Async +import Hasql.Session qualified as Session +import Hasql.Transaction qualified as Transaction +import Helpers.Scripts qualified as Scripts +import Helpers.Statements qualified as Statements +import Helpers.Transactions qualified as Transactions +import Prelude +import Test.Hspec + +spec :: SpecWith Scripts.ScopeParams +spec = do + it "Retrying transactions converge to a consistent balance under contention" \scopeParams -> + Scripts.onConnectionPair scopeParams \connection1 connection2 -> do + id1 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + id2 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + async1 <- Async.async (replicateM_ 1000 (Scripts.transaction connection1 (Transactions.transfer id1 id2 1))) + async2 <- Async.async (replicateM_ 1000 (Scripts.transaction connection2 (Transactions.transfer id1 id2 1))) + Async.wait async1 + Async.wait async2 + balance1 <- Scripts.session connection1 (Session.statement id1 Statements.getBalance) + balance2 <- Scripts.session connection1 (Session.statement id2 Statements.getBalance) + balance1 `shouldBe` Just 2000 + balance2 `shouldBe` Just (-2000) + + it "Non-retrying transactions fail with a serialization error under contention" \scopeParams -> + Scripts.onConnectionPair scopeParams \connection1 connection2 -> do + id1 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + id2 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + async1 <- Async.async (replicateM_ 1000 (Scripts.transactionNoRetry connection1 (Transactions.transfer id1 id2 1))) + async2 <- Async.async (replicateM_ 1000 (Scripts.transactionNoRetry connection2 (Transactions.transfer id1 id2 1))) + result1 <- Async.waitCatch async1 + result2 <- Async.waitCatch async2 + let serialError = sequenceA [result1, result2] + serialError `shouldSatisfy` either (isInfixOf "40001" . show) (const False) + + it "A concurrent read-only transaction does not lose updates from a writer" \scopeParams -> + Scripts.onConnectionPair scopeParams \connection1 connection2 -> do + id1 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + id2 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + async1 <- Async.async (replicateM_ 1000 (Scripts.transaction connection1 (Transactions.transfer id1 id2 1))) + async2 <- Async.async (replicateM_ 1000 (Scripts.transaction connection2 (Transaction.statement id1 Statements.getBalance))) + Async.wait async1 + Async.wait async2 + balance1 <- Scripts.session connection1 (Session.statement id1 Statements.getBalance) + balance2 <- Scripts.session connection1 (Session.statement id2 Statements.getBalance) + balance1 `shouldBe` Just 1000 + balance2 `shouldBe` Just (-1000) + + it "A transaction converges to a consistent balance alongside a concurrent bare session" \scopeParams -> + Scripts.onConnectionPair scopeParams \connection1 connection2 -> do + id1 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + id2 <- Scripts.session connection1 (Session.statement 0 Statements.createAccount) + async1 <- Async.async (Scripts.transaction connection1 (Transactions.transferTimes 200 id1 id2 1)) + async2 <- Async.async (Scripts.session connection2 (replicateM_ 200 (Session.statement (id1, 1) Statements.modifyBalance))) + Async.wait async1 + Async.wait async2 + balance1 <- Scripts.session connection1 (Session.statement id1 Statements.getBalance) + balance2 <- Scripts.session connection1 (Session.statement id2 Statements.getBalance) + balance1 `shouldBe` Just 400 + balance2 `shouldBe` Just (-200) diff --git a/src/integration-tests/Specs/SpecHook.hs b/src/integration-tests/Specs/SpecHook.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/SpecHook.hs @@ -0,0 +1,15 @@ +-- Docs: https://hspec.github.io/hspec-discover.html +module Specs.SpecHook where + +import Helpers.Adapters qualified as Adapters +import Helpers.Hooks qualified as Hooks +import Helpers.Scripts qualified as Scripts +import Test.Hspec + +hook :: SpecWith Scripts.ScopeParams -> Spec +hook hookedSpec = + Adapters.hook + ( aroundAllWith + (\action adapter -> Hooks.postgres17 \(host, port) -> action (adapter, host, port)) + hookedSpec + ) diff --git a/src/library/Hasql/Transaction/Private/Prelude.hs b/src/library/Hasql/Transaction/Private/Prelude.hs --- a/src/library/Hasql/Transaction/Private/Prelude.hs +++ b/src/library/Hasql/Transaction/Private/Prelude.hs @@ -62,6 +62,7 @@ import GHC.Generics as Exports (Generic, Generic1) import GHC.IO.Exception as Exports import Numeric as Exports +import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.)) import System.Environment as Exports import System.Exit as Exports import System.IO as Exports @@ -73,7 +74,6 @@ import Text.Printf as Exports (hPrintf, printf) import Text.Read as Exports (Read (..), readEither, readMaybe) import Unsafe.Coerce as Exports -import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.)) tryError :: (MonadError e m) => m a -> m (Either e a) tryError m =