diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.11.0.1 — 2026-10-20 + +### Bug Fixes + +- Preserve SQLSTATE and server message from `Tx.sql` script failures, matching + prepared statements: missing relations return `UnexpectedServerError "42P01"`, + while serialization failures and deadlocks return `TransientTransactionFailure`. +- Public reads using the shared pool helper now return existing typed server, + connection-loss and acquisition-timeout errors instead of flattened + `ConnectionError` diagnostics. Connection-session failures use `ConnectionLost`. + Callers matching only the old catch-all should handle the specific constructors. + No constructors, SQL work, or retry policy are added. + ## 0.11.0.0 — 2026-10-11 ### Breaking Changes diff --git a/kiroku-store.cabal b/kiroku-store.cabal --- a/kiroku-store.cabal +++ b/kiroku-store.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-store -version: 0.11.0.0 +version: 0.11.0.1 synopsis: High-performance PostgreSQL event store description: Kiroku is a PostgreSQL-backed event store for Haskell applications. It @@ -130,6 +130,7 @@ Test.PublisherIdleAdvance Test.PublisherRestartNoRebroadcast Test.ReadStream + Test.ScriptSessionErrors Test.StartupFailureSurfacing Test.StreamBridgeTermination Test.StreamHead diff --git a/src/Kiroku/Store/Effect.hs b/src/Kiroku/Store/Effect.hs --- a/src/Kiroku/Store/Effect.hs +++ b/src/Kiroku/Store/Effect.hs @@ -30,7 +30,6 @@ import Data.Map.Strict qualified as Map import Data.Maybe (isNothing) import Data.Text (Text) -import Data.Text qualified as T import Data.Time.Clock (UTCTime, getCurrentTime) import Data.UUID (UUID) import Data.UUID.V7 qualified as V7 @@ -51,7 +50,7 @@ import Hasql.Transaction.Sessions qualified as TxSessions import Kiroku.Store.Connection (KirokuStore (..)) import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore) -import Kiroku.Store.Error (StoreError (..), attributeMultiStreamError, emptyResultError, isTransientSerializationError, mapLinkUsageError, mapTransactionUsageError, mapUsageError, validateStreamName) +import Kiroku.Store.Error (StoreError (..), attributeMultiStreamError, emptyResultError, isTransientSerializationError, mapGenericUsageError, mapLinkUsageError, mapTransactionUsageError, mapUsageError, validateStreamName) import Kiroku.Store.HistoryRetention.Internal qualified as HistoryRetention import Kiroku.Store.HistoryRetention.Types import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop) @@ -527,7 +526,7 @@ usePool pool session = do result <- liftIO (Pool.use pool session) case result of - Left usageErr -> throwError (ConnectionError (T.pack (show usageErr))) + Left usageErr -> throwError (mapGenericUsageError usageErr) Right a -> pure a {- | Run a 'Tx.Transaction' against the pool using the supplied entry diff --git a/src/Kiroku/Store/Error.hs b/src/Kiroku/Store/Error.hs --- a/src/Kiroku/Store/Error.hs +++ b/src/Kiroku/Store/Error.hs @@ -190,7 +190,12 @@ Pattern matches on the error hierarchy: UsageError -> SessionUsageError -> StatementSessionError -> ServerStatementError -> ServerError + UsageError -> SessionUsageError -> ScriptSessionError -> ServerError +Scripts and statements share server classification. Connection establishment +and session connection loss use 'ConnectionLost'; acquisition timeout uses +'PoolAcquisitionTimeout'. Unsupported non-server errors retain 'ConnectionError'. + PostgreSQL error code mapping: 23505 (unique_violation) + events_pkey -> 'DuplicateEvent' 23505 (unique_violation) + stream_events_pkey -> 'DuplicateEvent' @@ -241,6 +246,8 @@ ConnectionLost (T.pack (show connErr)) AcquisitionTimeoutUsageError -> PoolAcquisitionTimeout + SessionUsageError (Errors.ConnectionSessionError detail) -> + ConnectionLost detail SessionUsageError sessionErr -> case extractServerError (SessionUsageError sessionErr) of Just (Errors.ServerError code message _ _ _) @@ -270,6 +277,10 @@ mapSessionError streamName expected = \case Errors.StatementSessionError _ _ _ _ _ stmtErr -> mapStatementError streamName expected stmtErr + Errors.ScriptSessionError _ serverErr -> + mapServerError streamName expected serverErr + Errors.ConnectionSessionError detail -> + ConnectionLost detail other -> ConnectionError ("Session error: " <> T.pack (show other)) @@ -519,6 +530,8 @@ -- | Walk a 'UsageError' down to the underlying 'Errors.ServerError', if any. extractServerError :: UsageError -> Maybe Errors.ServerError extractServerError = \case + SessionUsageError (Errors.ScriptSessionError _ serverErr) -> + Just serverErr SessionUsageError (Errors.StatementSessionError _ _ _ _ _ stmtErr) -> case stmtErr of Errors.ServerStatementError serverErr -> Just serverErr diff --git a/src/Kiroku/Store/Transaction.hs b/src/Kiroku/Store/Transaction.hs --- a/src/Kiroku/Store/Transaction.hs +++ b/src/Kiroku/Store/Transaction.hs @@ -94,8 +94,13 @@ existing transactional sites in 'Kiroku.Store.Append.appendMultiStream' and the hard-delete path. -Errors: a hasql 'Hasql.Pool.UsageError' is translated to -'Kiroku.Store.Error.ConnectionError'. Calling 'Tx.condemn' inside the +Errors from both 'Tx.sql' scripts and 'Tx.statement' preserve PostgreSQL +SQLSTATE and message through 'UnexpectedServerError' or +'TransientTransactionFailure'. Event-ID duplicates retain 'DuplicateEvent'; +opaque bodies do not fabricate stream context for other constraints. +Connection failures use 'ConnectionLost', and pool acquisition timeouts use +'PoolAcquisitionTimeout'. Unsupported driver errors retain 'ConnectionError'. +Calling 'Tx.condemn' inside the body marks the transaction for rollback at commit; the call returns the final value the body produced, but no SQL writes from this transaction will be visible. Mock interpreters of 'Store' reject this constructor. @@ -108,9 +113,11 @@ {- | Like 'runTransaction' but uses 'Hasql.Transaction.Sessions.transactionNoRetry' — the body is executed -exactly once even on PostgreSQL serialization conflicts. A conflict -surfaces as 'Kiroku.Store.Error.ConnectionError' carrying the underlying -'Hasql.Pool.UsageError'. +exactly once even on PostgreSQL serialization conflicts. Serialization +failures (@40001@) and deadlocks (@40P01@) surface as +'TransientTransactionFailure' with the original SQLSTATE and server message. +Use this entry point for a body that persistently raises those errors; the +retrying runner can keep re-executing it. -} runTransactionNoRetry :: (HasCallStack, Store :> es) => diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -53,6 +53,7 @@ import Test.PublisherIdleAdvance qualified as PublisherIdleAdvance import Test.PublisherRestartNoRebroadcast qualified as PublisherRestartNoRebroadcast import Test.ReadStream qualified as ReadStream +import Test.ScriptSessionErrors qualified as ScriptSessionErrors import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing import Test.StreamBridgeTermination qualified as StreamBridgeTermination import Test.StreamHead qualified as StreamHead @@ -95,6 +96,7 @@ HistoryRetention.spec HistoryRetentionMock.spec Transaction.spec + ScriptSessionErrors.spec TruncateBefore.spec ReadStream.spec StreamNameLookup.spec diff --git a/test/Test/ScriptSessionErrors.hs b/test/Test/ScriptSessionErrors.hs new file mode 100644 --- /dev/null +++ b/test/Test/ScriptSessionErrors.hs @@ -0,0 +1,143 @@ +module Test.ScriptSessionErrors (spec) where + +import Control.Exception (bracket) +import Control.Lens ((&), (.~)) +import Control.Monad (forM_) +import Data.Aeson qualified as Aeson +import Data.Generics.Labels () +import Data.Int (Int64) +import Data.Text (Text) +import Data.Text.Encoding qualified as TE +import Hasql.Connection.Settings qualified as Conn +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Errors qualified as Errors +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as Config +import Hasql.Session qualified as Session +import Hasql.Statement (preparable) +import Hasql.Transaction qualified as Tx +import Kiroku.Store +import Kiroku.Test.Postgres (withMigratedTestDatabase) +import System.Timeout (timeout) +import Test.Helpers (countEvents, makeEvent, withTestStore) +import Test.Hspec + +spec :: Spec +spec = describe "structured script session errors" $ do + around withTestStore $ do + it "preserves missing-relation SQLSTATE through runTransaction" $ \store -> do + result <- runStoreIO store (runTransaction (Tx.sql (TE.encodeUtf8 missingRelation))) + shouldBeMissingRelation result + it "preserves missing-relation SQLSTATE without retry" $ \store -> do + result <- runStoreIO store (runTransactionNoRetry (Tx.sql (TE.encodeUtf8 missingRelation))) + shouldBeMissingRelation result + forM_ ["40001", "40P01"] $ \code -> do + it ("preserves script SQLSTATE and message " <> show code) $ \store -> do + result <- timeout 5_000_000 $ runStoreIO store (runTransactionNoRetry (Tx.sql (TE.encodeUtf8 (raiseCode code)))) + result `shouldBe` Just (Left (TransientTransactionFailure code fixtureMessage)) + it ("preserves prepared-statement SQLSTATE and message " <> show code) $ \store -> do + result <- + timeout 5_000_000 $ + runStoreIO store $ + runTransactionNoRetry $ + Tx.statement () (preparable (raiseCode code) E.noParams D.noResult) + result `shouldBe` Just (Left (TransientTransactionFailure code fixtureMessage)) + it "rolls back application writes before a script failure" $ \store -> do + createSideTable store + result <- runStoreIO store $ runTransactionNoRetry $ do + Tx.sql "INSERT INTO public.script_error_rows VALUES (1)" + Tx.sql (TE.encodeUtf8 missingRelation) + shouldBeMissingRelation result + countSideRows store `shouldReturn` Right 0 + it "rolls back composed appends and application writes on script failure" $ \store -> do + createSideTable store + result <- runStoreIO store $ + runTransactionAppendingNoRetry (StreamName "script-rollback") NoStream [makeEvent "Created" (Aeson.object [])] $ \_ -> do + Tx.sql "INSERT INTO public.script_error_rows VALUES (1)" + Tx.sql (TE.encodeUtf8 missingRelation) + shouldBeMissingRelation result + countSideRows store `shouldReturn` Right 0 + runStoreIO store (getStream (StreamName "script-rollback")) `shouldReturn` Right Nothing + countEvents store `shouldReturn` 0 + + it "preserves missing-relation SQLSTATE through the public pool-backed read" $ + withMigratedTestDatabase $ \connStr -> + withStore (defaultConnectionSettings connStr) $ \store -> + withPool connStr [Config.initSession (Session.script "SET search_path TO pg_catalog")] $ \pool -> do + result <- runStoreIO (store & #pool .~ pool) (getStream (StreamName "missing")) + shouldBeMissingRelation result + + it "keeps genuine connection establishment failures typed for transactions and reads" $ + withTestStore $ \store -> + withPool "host=/nonexistent/kiroku-script-errors connect_timeout=1" [] $ \pool -> do + let broken = store & #pool .~ pool + tx <- timeout 5_000_000 $ runStoreIO broken (runTransactionNoRetry (Tx.sql "SELECT 1")) + tx `shouldSatisfy` isConnectionLost + readResult <- timeout 5_000_000 $ runStoreIO broken (getStream (StreamName "missing")) + readResult `shouldSatisfy` isConnectionLost + + it "routes script server errors through every contextual mapper" $ do + forM_ ["40001", "40P01", "42P01", "40003", "23503"] $ \code -> do + let usage = scriptUsage code fixtureMessage Nothing + expected = if code `elem` ["40001", "40P01"] then TransientTransactionFailure code fixtureMessage else UnexpectedServerError code fixtureMessage + mapGenericUsageError usage `shouldBe` expected + mapTransactionUsageError usage `shouldBe` expected + mapLinkUsageError (StreamName "target") usage `shouldBe` expected + isTransientSerializationError usage `shouldBe` (code `elem` ["40001", "40P01"]) + if code == "23503" + then mapUsageError "target" AnyVersion usage `shouldBe` StreamNotFound (StreamName "target") + else do + mapUsageError "target" AnyVersion usage `shouldBe` expected + attributeMultiStreamError [(StreamName "target", AnyVersion)] usage `shouldBe` expected + + it "retains script append and link constraint attribution" $ do + let usage = scriptUsage "23505" "duplicate key value violates unique constraint \"stream_events_pkey\"" Nothing + mapTransactionUsageError usage `shouldBe` DuplicateEvent Nothing + mapUsageError "target" AnyVersion usage `shouldBe` DuplicateEvent Nothing + mapLinkUsageError (StreamName "target") usage `shouldBe` EventAlreadyLinked (StreamName "target") Nothing + let named = scriptUsage "23505" "duplicate key value violates unique constraint \"ix_streams_stream_name\"" (Just "Key (stream_name)=(second) already exists.") + attributeMultiStreamError [(StreamName "first", AnyVersion), (StreamName "second", NoStream)] named + `shouldBe` StreamAlreadyExists (StreamName "second") + + it "distinguishes session connection loss, acquisition timeout and unsupported errors" $ do + let lost = Pool.SessionUsageError (Errors.ConnectionSessionError "connection detail") + fallback = Pool.SessionUsageError (Errors.DriverSessionError "driver detail") + forM_ [mapGenericUsageError, mapTransactionUsageError, mapUsageError "target" AnyVersion, mapLinkUsageError (StreamName "target")] $ \mapper -> do + mapper lost `shouldBe` ConnectionLost "connection detail" + mapper Pool.AcquisitionTimeoutUsageError `shouldBe` PoolAcquisitionTimeout + mapper fallback `shouldSatisfy` (\case ConnectionError _ -> True; _ -> False) + isTransientSerializationError lost `shouldBe` False + +missingRelation :: Text +missingRelation = "INSERT INTO public.kiroku_script_missing_relation VALUES (1)" + +fixtureMessage :: Text +fixtureMessage = "script error fixture" + +raiseCode :: Text -> Text +raiseCode code = "DO $$ BEGIN RAISE EXCEPTION USING ERRCODE = '" <> code <> "', MESSAGE = '" <> fixtureMessage <> "'; END $$" + +scriptUsage :: Text -> Text -> Maybe Text -> Pool.UsageError +scriptUsage code message detail = Pool.SessionUsageError (Errors.ScriptSessionError "fixture SQL" (Errors.ServerError code message detail Nothing Nothing)) + +shouldBeMissingRelation :: (Show a) => Either StoreError a -> Expectation +shouldBeMissingRelation result = case result of + Left (UnexpectedServerError "42P01" _) -> pure () + other -> expectationFailure ("Expected structured 42P01, got " <> show other) + +isConnectionLost :: Maybe (Either StoreError a) -> Bool +isConnectionLost (Just (Left (ConnectionLost _))) = True +isConnectionLost _ = False + +withPool :: Text -> [Config.Setting] -> (Pool.Pool -> IO a) -> IO a +withPool connStr extra = bracket (Pool.acquire (Config.settings ([Config.staticConnectionSettings (Conn.connectionString connStr), Config.size 1, Config.acquisitionTimeout 1] <> extra))) Pool.release + +createSideTable :: KirokuStore -> IO () +createSideTable store = runStoreIO store (runTransactionNoRetry (Tx.sql "CREATE TABLE public.script_error_rows (id integer)")) `shouldReturn` Right () + +countSideRows :: KirokuStore -> IO (Either StoreError Int64) +countSideRows store = + runStoreIO store $ + runTransactionNoRetry $ + Tx.statement () (preparable "SELECT count(*) FROM public.script_error_rows" E.noParams (D.singleRow (D.column (D.nonNullable D.int8))))