diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## 0.6.0.0 — 2026-08-12 + +### Breaking Changes + +* The exported `Store` effect gains `GetVisibleGlobalHeadPosition`. Exhaustive + custom and mock interpreters must handle the new constructor. + +### New Features + +* `Kiroku.Store.Read.visibleGlobalHeadPosition` returns the greatest global + position still visible in `$all`, or zero when no event remains, through a + payload-free scalar query. Hard deletion of the visible tail can make the + result regress while the authoritative append frontier remains monotonic; no + event payload is read and the configured decode hook is not invoked. + ## 0.5.0.0 — 2026-08-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.5.0.0 +version: 0.6.0.0 synopsis: High-performance PostgreSQL event store description: Kiroku is a PostgreSQL-backed event store for Haskell applications. It @@ -127,6 +127,8 @@ Test.SubscriptionState Test.Transaction Test.TruncateBefore + Test.VisibleGlobalHeadPosition + Test.VisibleGlobalHeadPositionMock ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: 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 @@ -78,6 +78,14 @@ ReadStreamBackward :: StreamName -> StreamVersion -> Int32 -> Store m (Vector RecordedEvent) ReadAllForward :: GlobalPosition -> Int32 -> Store m (Vector RecordedEvent) ReadAllBackward :: GlobalPosition -> Int32 -> Store m (Vector RecordedEvent) + {- | Read the greatest position whose @$all@ junction still survives, or + zero when no event remains visible. Unlike + 'Kiroku.Store.Subscription.subscriptionCheckpointInventory', this is a + visibility cursor rather than the authoritative append frontier. + + Surfaced as 'Kiroku.Store.Read.visibleGlobalHeadPosition'. + -} + GetVisibleGlobalHeadPosition :: Store m GlobalPosition GetStream :: StreamName -> Store m (Maybe StreamInfo) {- | Resolve a 'StreamName' to its surrogate 'StreamId' without materializing the full 'StreamInfo' row. Mirrors 'GetStream'\'s @@ -234,6 +242,9 @@ usePool (store ^. #pool) $ Session.statement (cursor, limit) SQL.readAllBackwardStmt liftIO $ decodeEvents (store ^. #storeSettings) evs + GetVisibleGlobalHeadPosition -> + usePool (store ^. #pool) $ + Session.statement () SQL.visibleGlobalHeadPositionStmt GetStream (StreamName name) -> usePool (store ^. #pool) $ Session.statement name SQL.getStreamStmt diff --git a/src/Kiroku/Store/Read.hs b/src/Kiroku/Store/Read.hs --- a/src/Kiroku/Store/Read.hs +++ b/src/Kiroku/Store/Read.hs @@ -4,6 +4,7 @@ readStreamBackward, readAllForward, readAllBackward, + visibleGlobalHeadPosition, readCategory, getStream, lookupStreamId, @@ -135,6 +136,25 @@ Int32 -> Eff es (Vector RecordedEvent) readAllBackward startPos limit = send (ReadAllBackward startPos limit) + +{- | Return the greatest 'GlobalPosition' still visible in the global @$all@ +stream, or @'GlobalPosition' 0@ when no event remains. + +Logical truncation and soft deletion leave @$all@ junctions intact, so neither +changes this value. Hard deletion removes junctions and can make the visible +head move backward or return to zero. This differs from +'Kiroku.Store.Subscription.Types.storePosition' in +'Kiroku.Store.Subscription.Types.SubscriptionCheckpointInventory', which is +the authoritative append frontier and remains monotonic after hard deletion. + +The value is observed by one database statement. A concurrent append or hard +deletion may change the visible head immediately after this function returns; +the result does not retain a database snapshot. +-} +visibleGlobalHeadPosition :: + (HasCallStack, Store :> es) => + Eff es GlobalPosition +visibleGlobalHeadPosition = send GetVisibleGlobalHeadPosition {- | Read events whose source stream's category prefix matches the given 'CategoryName', in 'GlobalPosition' order. diff --git a/src/Kiroku/Store/SQL.hs b/src/Kiroku/Store/SQL.hs --- a/src/Kiroku/Store/SQL.hs +++ b/src/Kiroku/Store/SQL.hs @@ -21,6 +21,7 @@ eventExistsInStreamStmt, lookupStreamNamesStmt, currentGlobalPositionStmt, + visibleGlobalHeadPositionStmt, -- * Consumer-group read statements readCategoryForwardConsumerGroupStmt, @@ -456,6 +457,22 @@ "SELECT stream_version FROM streams WHERE stream_id = 0" E.noParams (D.singleRow (D.column (D.nonNullable D.int8))) + +-- | Read the greatest position still visible in the global $all stream. +visibleGlobalHeadPositionStmt :: Statement () GlobalPosition +visibleGlobalHeadPositionStmt = + preparable + """ + SELECT COALESCE(( + SELECT stream_version + FROM stream_events + WHERE stream_id = 0 + ORDER BY stream_version DESC + LIMIT 1 + ), 0) + """ + E.noParams + (D.singleRow (GlobalPosition <$> D.column (D.nonNullable D.int8))) -- | Get stream metadata by name. getStreamStmt :: Statement Text (Maybe StreamInfo) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -63,6 +63,8 @@ import Test.SubscriptionState qualified as SubscriptionState import Test.Transaction qualified as Transaction import Test.TruncateBefore qualified as TruncateBefore +import Test.VisibleGlobalHeadPosition qualified as VisibleGlobalHeadPosition +import Test.VisibleGlobalHeadPositionMock qualified as VisibleGlobalHeadPositionMock main :: IO () main = withSharedMigratedPostgres $ hspec $ do @@ -102,6 +104,8 @@ SubscriptionRegistry.spec SubscriptionRetryDeadLetter.spec EventTypeFilter.spec + VisibleGlobalHeadPosition.spec + VisibleGlobalHeadPositionMock.spec around withTestStore $ do describe "schema migrations" $ do it "installs every Kiroku table under the kiroku schema" $ \store -> do diff --git a/test/Test/PerformanceStructure.hs b/test/Test/PerformanceStructure.hs --- a/test/Test/PerformanceStructure.hs +++ b/test/Test/PerformanceStructure.hs @@ -53,6 +53,15 @@ queryPlanSpec = describe "production query plans" $ aroundAll withQueryPlanStore $ do + it "visible global head lookup uses ux_stream_events_stream_version without Sort" $ \store -> do + plan <- + explainProductionStatement + store + SQL.visibleGlobalHeadPositionStmt + [] + expectIndex "ux_stream_events_stream_version" plan + expectNoNodeType "Sort" plan + it "category high-cursor reads use ix_stream_events_all_by_origin" $ \store -> do plan <- explainProductionStatement diff --git a/test/Test/VisibleGlobalHeadPosition.hs b/test/Test/VisibleGlobalHeadPosition.hs new file mode 100644 --- /dev/null +++ b/test/Test/VisibleGlobalHeadPosition.hs @@ -0,0 +1,105 @@ +{-# LANGUAGE TypeApplications #-} + +module Test.VisibleGlobalHeadPosition (spec) where + +import Control.Lens ((&), (.~), (^.)) +import Data.Aeson qualified as Aeson +import Data.Generics.Labels () +import Data.IORef (modifyIORef', newIORef, readIORef) +import Effectful (runEff) +import Effectful.Error.Static (runErrorNoCallStack) +import Kiroku.Store +import Test.Helpers (makeEvent, withTestStore, withTestStoreSettings) +import Test.Hspec + +spec :: Spec +spec = describe "visible global head position" $ do + it "returns zero for an empty migrated store through both public runners" $ + withTestStore $ \store -> do + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition 0) + + resourceResult <- + runEff + . runErrorNoCallStack @StoreError + . runKirokuStoreWith store + . runStoreResource + $ visibleGlobalHeadPosition + resourceResult `shouldBe` Right (GlobalPosition 0) + + it "returns the greatest appended position" $ + withTestStore $ \store -> do + appendEvents store (StreamName "visible-head-populated") 3 + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition 3) + + it "falls back across hard-deleted tails while the append frontier stays monotonic" $ + withTestStore $ \store -> do + let first = StreamName "visible-head-first" + middle = StreamName "visible-head-middle" + lastStream = StreamName "visible-head-last" + appendEvents store first 1 + appendEvents store middle 1 + appendEvents store lastStream 1 + + Right (Just _) <- runStoreIO store $ hardDeleteStream middle + assertHeadAndFrontier store 3 3 + + Right (Just _) <- runStoreIO store $ hardDeleteStream lastStream + assertHeadAndFrontier store 1 3 + + Right (Just _) <- runStoreIO store $ hardDeleteStream first + assertHeadAndFrontier store 0 3 + + it "keeps logically truncated and soft-deleted events visible in $all" $ + withTestStore $ \store -> do + let name = StreamName "visible-head-logical-lifecycle" + appendEvents store name 3 + + Right (Just _) <- + runStoreIO store $ + setStreamTruncateBefore name (StreamVersion 3) + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition 3) + + Right (Just _) <- runStoreIO store $ softDeleteStream name + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition 3) + + it "does not invoke the event decode hook" $ do + decodeCalls <- newIORef (0 :: Int) + let failingHook event = do + modifyIORef' decodeCalls (+ 1) + ioError (userError ("unexpected decode of " <> show (event ^. #eventId))) + tweak settings = + settings + & #storeSettings + .~ defaultStoreSettings{decodeHook = Just failingHook} + withTestStoreSettings tweak $ \store -> do + appendEvents store (StreamName "visible-head-no-decode") 1 + + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition 1) + readIORef decodeCalls `shouldReturn` 0 + +appendEvents :: KirokuStore -> StreamName -> Int -> IO () +appendEvents store name count = do + let events = + [ makeEvent "VisibleHeadEvent" (Aeson.object [("ordinal", Aeson.Number (fromIntegral ordinal))]) + | ordinal <- [1 .. count] + ] + result <- runStoreIO store $ appendToStream name NoStream events + case result of + Left err -> expectationFailure ("append failed: " <> show err) + Right _ -> pure () + +assertHeadAndFrontier :: KirokuStore -> Integer -> Integer -> IO () +assertHeadAndFrontier store expectedHead expectedFrontier = do + runStoreIO store visibleGlobalHeadPosition + `shouldReturn` Right (GlobalPosition (fromIntegral expectedHead)) + result <- runStoreIO store subscriptionCheckpointInventory + case result of + Left err -> expectationFailure ("inventory read failed: " <> show err) + Right inventory -> + inventory ^. #storePosition + `shouldBe` GlobalPosition (fromIntegral expectedFrontier) diff --git a/test/Test/VisibleGlobalHeadPositionMock.hs b/test/Test/VisibleGlobalHeadPositionMock.hs new file mode 100644 --- /dev/null +++ b/test/Test/VisibleGlobalHeadPositionMock.hs @@ -0,0 +1,31 @@ +module Test.VisibleGlobalHeadPositionMock (spec) where + +import Control.Monad.IO.Class (liftIO) +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Effectful (Eff, IOE, runEff, (:>)) +import Effectful.Dispatch.Dynamic (interpret_) +import Kiroku.Store.Effect (Store (..)) +import Kiroku.Store.Read (visibleGlobalHeadPosition) +import Kiroku.Store.Types (GlobalPosition (..)) +import Test.Hspec + +spec :: Spec +spec = describe "visible global head position mock" $ + it "returns the configured position through one Store effect call" $ do + calls <- newIORef (0 :: Int) + let expected = GlobalPosition 23 + actual <- runEff $ runVisibleHeadMock calls expected visibleGlobalHeadPosition + actual `shouldBe` expected + readIORef calls `shouldReturn` 1 + +runVisibleHeadMock :: + (IOE :> es) => + IORef Int -> + GlobalPosition -> + Eff (Store : es) a -> + Eff es a +runVisibleHeadMock calls expected = interpret_ $ \case + GetVisibleGlobalHeadPosition -> do + liftIO $ modifyIORef' calls (+ 1) + pure expected + _ -> error "unexpected Store operation in visible-head mock"