kiroku-store 0.2.0.0 → 0.3.0.0

raw patch · 35 files changed

+2779/−438 lines, 35 filesdep +mtldep ~shibuya-corePVP ok

version bump matches the API change (PVP)

Dependencies added: mtl

Dependency ranges changed: shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +201/−0
• bench/Main.hs +150/−2
• bench/ShibuyaOverhead.hs +5/−4
• kiroku-store.cabal +30/−18
• src/Kiroku/Store/Append.hs +8/−7
• src/Kiroku/Store/Effect.hs +155/−53
• src/Kiroku/Store/Error.hs +140/−6
• src/Kiroku/Store/Lifecycle.hs +42/−1
• src/Kiroku/Store/Link.hs +12/−3
• src/Kiroku/Store/Notification.hs +36/−32
• src/Kiroku/Store/Observability.hs +37/−15
• src/Kiroku/Store/Read.hs +38/−13
• src/Kiroku/Store/SQL.hs +168/−36
• src/Kiroku/Store/Subscription.hs +97/−72
• src/Kiroku/Store/Subscription/EventPublisher.hs +74/−29
• src/Kiroku/Store/Subscription/Fsm.hs +1/−1
• src/Kiroku/Store/Subscription/Stream.hs +68/−14
• src/Kiroku/Store/Subscription/Types.hs +23/−16
• src/Kiroku/Store/Subscription/Worker.hs +78/−23
• src/Kiroku/Store/Transaction.hs +56/−15
• src/Kiroku/Store/Types.hs +52/−3
• test/Main.hs +182/−3
• test/Test/Category.hs +70/−0
• test/Test/Concurrency.hs +31/−0
• test/Test/Helpers.hs +40/−0
• test/Test/NotifyGuard.hs +122/−0
• test/Test/PublisherCallbackResilience.hs +106/−0
• test/Test/PublisherIdleAdvance.hs +140/−0
• test/Test/ReadStream.hs +128/−69
• test/Test/StartupFailureSurfacing.hs +81/−0
• test/Test/StreamBridgeTermination.hs +108/−0
• test/Test/StreamNameLookup.hs +29/−3
• test/Test/SubscriptionRegistry.hs +48/−0
• test/Test/Transaction.hs +98/−0
• test/Test/TruncateBefore.hs +125/−0

@@ -2,6 +2,207 @@  ## Unreleased +## 0.3.0.0 — 2026-07-11++This release requires migrations `0003`–`0007` from `kiroku-store-migrations`.++### Breaking Changes++* Backward reads now paginate. `readStreamBackward` and `readAllBackward` treat a+  nonzero cursor as an *exclusive upper bound*, so a page returns events older+  than the cursor instead of newer ones. The interpreter maps the caller cursor+  `0` to `maxBound`, so `StreamVersion 0` / `GlobalPosition 0` still means "start+  from the latest event" (versions and positions start at 1, so the sentinel can+  never collide with a real value). Previously a nonzero cursor read in the wrong+  direction, which made backward pagination impossible: feeding the last returned+  version back in returned the same newest page again. Any caller passing a+  nonzero backward cursor gets different results, and any mock `Store`+  interpreter must mirror the `0 -> maxBound` mapping.+* Oversized stream names are rejected before any database work. `StoreError`+  gains `StreamNameTooLong !StreamName !Int` (the `Int` is the offending UTF-8+  byte length), and `Kiroku.Store.Error` exports `maxStreamNameBytes` (512) and+  `validateStreamName :: StreamName -> Either StoreError ()`. Every+  application-stream entry point — `appendToStream`, `appendMultiStream`,+  `linkToStream`, `softDeleteStream`, `hardDeleteStream`, `undeleteStream`, and+  the `runTransactionAppending*` wrappers — runs the same validation (reserved+  `$all` plus the length bound) up front instead of only rejecting `$all`. The+  bound exists because the append notification payload embeds the stream name and+  PostgreSQL rejects `pg_notify` payloads of 8,000 bytes or more; a name that+  used to fail late as an opaque trigger error now fails immediately with a typed+  error. Exhaustive matches on `StoreError` must handle the new constructor.+* Publisher queues are created only for non-group `AllStreams` subscriptions.+  `Category` subscriptions and all consumer-group members no longer register a+  bounded queue with the `EventPublisher`; they are DB-driven in live mode. As a+  consequence `SubscriptionConfig.queueCapacity` and+  `SubscriptionConfig.overflowPolicy` have no effect for those targets, and they+  can no longer be paused, dropped, or terminated with `SubscriptionOverflowed`.+  In the exposed `Kiroku.Store.Subscription.Worker` module, `runWorker`'s+  signature changed from taking a `TBQueue (Vector RecordedEvent)` plus a `TVar+  SubscriberStatus` to taking a single new exported sum `LiveSource+  (LiveFromPublisherQueue | LiveFromCategoryNotify | LiveFromGroupPolling)`,+  fixed at `subscribe` time from the config's `(consumerGroup, target)` shape.+* Empty append batches are rejected. `StoreError` gains `EmptyAppendBatch+  !StreamName` and `AppendConflict` gains `EmptyAppendBatchConflict !StreamName`+  (mapped by `appendConflictToStoreError`). `appendToStream` with `[]` now fails+  with `EmptyAppendBatch` before any pool work; previously it took the global+  `$all` row lock, fired NOTIFY triggers, created an empty stream under+  `NoStream`, and surfaced a misleading `WrongExpectedVersion` /+  `StreamNotFound`. `appendMultiStream []` is an explicit no-op returning `[]`+  without a round trip, while a per-stream empty event list inside a non-empty op+  list is rejected with `EmptyAppendBatch`.+* Link failures are typed. `StoreError` gains `EventAlreadyLinked !StreamName+  !(Maybe EventId)` (the `23505` unique violation on `stream_events_pkey`,+  carrying the offending event id when PostgreSQL's detail string is parseable)+  and `LinkSourceEventMissing !StreamName` (the `23502` not-null violation raised+  when `linkToStream` references an event id that does not exist; the whole batch+  rolls back). Both previously surfaced as an opaque `ConnectionError` carrying a+  stringified `UsageError`.+* `StreamInfo` gains a `truncateBefore :: !StreamVersion` field. Code+  constructing or exhaustively matching `StreamInfo` must be updated.+* The `Store` effect gains `SetStreamTruncateBefore :: StreamName ->+  StreamVersion -> Store m (Maybe StreamId)` and `EventExistsInStream ::+  StreamName -> EventId -> Store m Bool`. Mock and test interpreters of the+  `Store` effect must add arms for both.+* `Kiroku.Store.SQL`'s hard-delete statements were resplit.+  `deleteStreamJunctionsStmt` is removed, replaced by `deleteAllRowsForOriginStmt`,+  `deleteJunctionsByEventIdsStmt`, `deleteStreamOwnJunctionsStmt`, and+  `deleteDeadLettersForOrphanedEventsStmt`, which the interpreter runs in sequence+  before `deleteOrphanedEventsStmt`.+* Checkpoint-load failures now fail the subscription loudly. A `UsageError` while+  reading a subscription's saved checkpoint at startup surfaces through the+  handle's `wait` as `Left e` and the worker stops; it no longer falls back to+  `GlobalPosition 0` and silently re-processes the whole history.+* `subscriptionAckStream` and `subscriptionStream` reject a zero buffer, throwing+  the new `InvalidStreamBufferSize` exception (exported from+  `Kiroku.Store.Subscription.Stream`) when `bufferSize` is `0`; a zero-capacity+  `TBQueue` would deadlock the bridge handler on its first delivery.+* Duplicate events inside transaction bodies are typed. `runTransaction`,+  `runTransactionNoRetry`, and the `runTransactionAppending*` wrappers map a+  `23505` violation on `events_pkey` / `stream_events_pkey` to `DuplicateEvent+  (Just eid)` (new `mapTransactionUsageError` in `Kiroku.Store.Error`) instead of+  the previous opaque `ConnectionError`.++### New Features++* Logical truncate-before marker (close-the-book compaction).+  `Kiroku.Store.Lifecycle.setStreamTruncateBefore :: StreamName -> StreamVersion+  -> Eff es (Maybe StreamId)` and `clearStreamTruncateBefore :: StreamName -> Eff+  es (Maybe StreamId)` set a per-stream cursor below which ordered per-stream+  reads hide events. `readStreamForward`, `readStreamBackward`, and the paged+  `readStreamForwardStream` return only events whose `streamVersion` is at or+  above the marker; the `$all` global log, `readCategory`, subscriptions, and+  existence probes are deliberately unaffected, so global history stays complete+  and future projections can still be built from it. No events are deleted and+  the operation is fully reversible. The typical use is snapshot-and-compact:+  append a snapshot event at version `V`, then call `setStreamTruncateBefore name+  V` so rehydration starts there. The marker is surfaced on+  `StreamInfo.truncateBefore` via `getStream`, is idempotent, returns `Nothing`+  for a missing or soft-deleted stream, and rejects `$all` with+  `ReservedStreamName`. Backed by the new `streams.truncate_before` column+  shipped by `kiroku-store-migrations`.+* `Kiroku.Store.Read.eventExistsInStream` adds an indexed point lookup for+  idempotency checks that need to know whether a specific event id is present in+  a live stream without scanning that stream. Soft-deleted streams report+  `False`, matching `readStreamForward` visibility.+* `Kiroku.Store.Types` exports `categoryName :: StreamName -> CategoryName` (the+  substring before the first `-`) and its dual `streamNameInCategory ::+  CategoryName -> Text -> StreamName`. This is the canonical Haskell mirror of+  the `streams.category` generated column and of the rule used by `readCategory`+  and category-targeted subscriptions; previously every consumer hand-rolled it.+  `Notification.categoryFromPayload` now routes through `categoryName`, so the+  rule has one definition.+* New observability event `KirokuEventPublisherLoopError !SomeException`, emitted+  when the publisher's broadcast iteration throws a synchronous non-`UsageError`+  exception (typically a user-supplied `decodeHook` or observability handler).+  Exhaustive matches on `KirokuEvent` must handle it.++### Bug Fixes++* The publisher survives failing callbacks. A synchronous exception from the+  broadcast iteration is caught, reported as `KirokuEventPublisherLoopError`, and+  retried on the next notification or the 30-second safety poll without advancing+  `lastPublished`; previously it killed the publisher loop and stalled every live+  subscription. All store-internal threads (publisher, worker, notifier) now emit+  observability events through a hardened `emitOrDrop` that drops synchronous+  exceptions thrown by the handler and rethrows async ones, so a throwing+  metrics or logging callback can no longer kill a store thread.+* Streamly bridges terminate with the worker's outcome. `subscriptionStream` and+  `subscriptionAckStream` end the stream normally when the worker stops cleanly+  or is cancelled, and rethrow the worker's exception from the next stream pull+  when it dies for any other reason (overflow, handler exception, dead-letter+  database error, decode-hook exception); previously the consumer blocked+  forever. The cancel action no longer writes a sentinel into the bounded queue,+  so it cannot block behind a full bridge buffer.+* Subscription startup no longer leaks registrations. The pre-fork window in+  `subscribe` is masked and each acquired registration is paired with+  `bracketOnError` cleanup, so an exception between publisher registration and+  the worker fork can no longer leave a subscriber registered with no reader.+  Notifier startup releases its connection if `LISTEN` fails.+* Hard delete purges dead letters. `hardDeleteStream` deletes+  `kiroku.dead_letters` rows for exactly the events whose payloads it orphans,+  before deleting those payloads. Because `dead_letters.event_id` has a foreign+  key to `events(event_id)` with no cascade, hard-deleting a stream that had+  dead-lettered events previously failed on that constraint.+* Single-stream appends retry transient conflicts. `appendToStream` retries once+  when the first attempt fails with PostgreSQL `40001` (serialization failure) or+  `40P01` (deadlock detected) — the same SQLSTATE set `hasql-transaction`+  retries. Event ids are prepared before the first attempt, so the retry is+  idempotent.+* `appendToStream` and `appendMultiStream` surface duplicate events raised from+  inside transaction bodies as `DuplicateEvent` rather than a stringified+  connection error.++### Performance++* `appendMultiStream` now pipelines its deterministic pre-lock and per-stream+  append statements through Hasql pipeline mode, while preserving all-or-nothing+  rollback and transient deadlock retry behavior. The local benchmark regression+  gate measured `All.reliability-audit.appendMultiStream 3 existing streams` at+  268 us, 29% faster than the refreshed baseline.+* The publisher does no row fetches while idle. With an empty subscriber registry+  the publisher loop advances `lastPublished` from the `$all` tail with a+  single-row query instead of fetching and decoding full event rows, so a store+  whose only subscriptions are `Category` or consumer-group members pays no+  fan-out cost. It re-checks registry emptiness in the same STM transaction as+  the position write, and offers an in-flight batch to subscribers that+  registered after the snapshot, so a subscriber attaching mid-tick never skips+  events.+* No empty read round trips. `readStreamForwardStream` stops after a page shorter+  than `pageSize` rather than always probing for one more empty page (an exact+  multiple of `pageSize` still costs one final probe), flattens pages without an+  intermediate list, and `lookupStreamNames []` short-circuits to an empty map in+  the interpreter with no database round trip.+* Index hygiene and `streams` fillfactor tuning (migration `0005`).++### Other Changes++* `GlobalPosition`'s documented contract is weakened to "strictly increasing and+  totally ordered" — nothing more. Treat values as opaque cursors: construct one+  only from `0` or from a value this store returned, never by arithmetic, and+  never assume density (`pos + 1` may not exist). The current implementation+  still assigns contiguous positions, but contiguity is now explicitly an+  implementation detail, not an API guarantee.+* `linkToStream` is marked provisional. It is the only public feature that+  requires the `stream_events` junction-table layout, which a future+  global-position migration may replace; it may be removed or redesigned.+* `RetryPolicy.retryMaxAttempts` is documented as a bound on total deliveries,+  not redeliveries: the first delivery counts as attempt 1, so+  `defaultRetryPolicy` (5) means the first delivery plus four redeliveries before+  `DeadLetterMaxAttempts`. Behavior is unchanged; the previous haddock was wrong.+* `appendToStreamTx` and the `runTransactionAppending*` wrappers document the+  `$all` lock-hold hazard: the append updates Kiroku's global `$all` row and+  PostgreSQL holds that row lock until the transaction ends, so every statement+  the continuation runs afterwards blocks all other appends store-wide. Keep+  continuations minimal and run append-independent work first.+* `WrongExpectedVersion` and `WrongExpectedVersionConflict`'s third field is+  documented as a placeholder: the append statement returns zero rows on a+  mismatch and the store does not issue an extra read, so it is always+  `StreamVersion 0`. Callers needing the live version must use `getStream`.+* The append NOTIFY trigger is now split into INSERT/UPDATE triggers that exclude+  the internal `$all` row and only fire when `stream_version` actually advances,+  so soft-delete and undelete no longer emit spurious append notifications+  (migration `0003`).+ ## 0.2.0.0 — 2026-05-31  ### Breaking Changes

@@ -3,13 +3,18 @@  module Main where +import Control.Concurrent (threadDelay) import Control.Concurrent.Async (mapConcurrently_)+import Control.Concurrent.Async qualified as Async+import Control.Exception (finally) import Control.Lens ((^.))+import Control.Monad (replicateM_) import Data.Aeson qualified as Aeson import Data.Functor.Contravariant ((>$<)) import Data.Generics.Labels () import Data.IORef import Data.Int (Int64)+import Data.Maybe (isNothing) import Data.Text (Text) import Data.Text qualified as T import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime)@@ -20,12 +25,16 @@ import GHC.Generics (Generic) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E+import Hasql.Pipeline qualified as Pipeline import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session-import Hasql.Statement (Statement, preparable)+import Hasql.Statement (Statement, preparable, unpreparable) import Hasql.Transaction qualified as Tx import Hasql.Transaction.Sessions qualified as TxSessions import Kiroku.Store+import Kiroku.Store.Effect (buildAppendParams, prepareEvents)+import Kiroku.Store.SQL qualified as SQL+import Kiroku.Test.Postgres (migrateTestDatabase) import Test.Tasty.Bench  data RawAppendParams = RawAppendParams@@ -98,6 +107,45 @@         rawProductionAppendParamsEncoder         rawAppendResultDecoder +beginStmt :: Statement () ()+beginStmt = unpreparable "BEGIN" E.noParams D.noResult++commitStmt :: Statement () ()+commitStmt = unpreparable "COMMIT" E.noParams D.noResult++rollbackStmt :: Statement () ()+rollbackStmt = unpreparable "ROLLBACK" E.noParams D.noResult++pipelinedMultiAppend4Streams :: [(StreamName, Text)]+pipelinedMultiAppend4Streams =+    namedBenchStreams "pipe4" 4++pipelinedMultiAppend8Streams :: [(StreamName, Text)]+pipelinedMultiAppend8Streams =+    namedBenchStreams "pipe8" 8++pipelinedContentionCurrentGroups :: [[(StreamName, Text)]]+pipelinedContentionCurrentGroups =+    namedBenchGroups "pipe-current-writer" 4 4++pipelinedContentionPipelinedGroups :: [[(StreamName, Text)]]+pipelinedContentionPipelinedGroups =+    namedBenchGroups "pipe-pipelined-writer" 4 4++namedBenchGroups :: Text -> Int -> Int -> [[(StreamName, Text)]]+namedBenchGroups prefix groups streamsPerGroup =+    [ namedBenchStreams (prefix <> "-" <> T.pack (show groupId)) streamsPerGroup+    | groupId <- [1 .. groups]+    ]++namedBenchStreams :: Text -> Int -> [(StreamName, Text)]+namedBenchStreams prefix count =+    [ ( StreamName (prefix <> "-" <> T.pack (show i))+      , prefix <> "-event-" <> T.pack (show i)+      )+    | i <- [1 .. count]+    ]+ rawScalarAppendAnyVersionSQL :: Text rawScalarAppendAnyVersionSQL =     """@@ -597,6 +645,58 @@                 ]     forceAppendList r +runCurrentMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO ()+runCurrentMultiAppend store streams = do+    r <-+        runStoreIO store $+            appendMultiStream+                [ (sn, AnyVersion, [makeEvent typ])+                | (sn, typ) <- streams+                ]+    forceAppendList r++runPipelinedMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO ()+runPipelinedMultiAppend store streams = do+    now <- getCurrentTime+    params <-+        mapM+            ( \(StreamName name, typ) -> do+                prepared <- prepareEvents [makeEvent typ]+                pure (buildAppendParams name now prepared)+            )+            streams+    let names = V.fromList [name | (StreamName name, _) <- streams]+    result <-+        Pool.use (store ^. #pool) $ do+            results <-+                Session.pipeline $+                    Pipeline.statement () beginStmt+                        *> Pipeline.statement names SQL.lockStreamsForMultiStmt+                        *> traverse (`Pipeline.statement` SQL.appendAnyVersion) params+            Session.statement () $+                if any isNothing results+                    then rollbackStmt+                    else commitStmt+            pure results+    forcePipelinedAppendList result++runSingleAppendUnderMultiWriters ::+    KirokuStore ->+    IORef Int ->+    (KirokuStore -> [(StreamName, Text)] -> IO ()) ->+    Text ->+    [[(StreamName, Text)]] ->+    IO ()+runSingleAppendUnderMultiWriters store runCounter multiAppendRunner prefix groups = do+    runId <- atomicModifyIORef' runCounter (\m -> (m + 1, m))+    workers <- mapM (Async.async . replicateM_ 20 . multiAppendRunner store) groups+    let cleanup = mapM_ Async.cancel workers+        appendMeasured i = do+            let sn = StreamName (prefix <> "-single-" <> T.pack (show runId) <> "-" <> T.pack (show i))+            r <- runStoreIO store $ appendToStream sn AnyVersion [makeEvent (prefix <> "-single")]+            forceAppend r+    (threadDelay 20_000 >> mapM_ appendMeasured [1 .. 10 :: Int] >> mapM_ Async.wait workers) `finally` cleanup+ -- | Exercise subscription catch-up over a compact category-local backlog. runSubscriptionCatchup :: KirokuStore -> IORef Int -> IO () runSubscriptionCatchup store runCounter = do@@ -641,6 +741,16 @@ forceAppendList (Right rs) = mapM_ (\r -> (r ^. #streamVersion) `seq` (r ^. #globalPosition) `seq` pure ()) rs forceAppendList (Left e) = error ("Benchmark appendMultiStream failed: " <> show e) +forcePipelinedAppendList :: Either Pool.UsageError [Maybe AppendResult] -> IO ()+forcePipelinedAppendList (Right rs) =+    mapM_+        ( \mResult -> case mResult of+            Just r -> (r ^. #streamVersion) `seq` (r ^. #globalPosition) `seq` pure ()+            Nothing -> error "Pipelined appendMultiStream benchmark produced no result"+        )+        rs+forcePipelinedAppendList (Left e) = error ("Pipelined appendMultiStream benchmark failed: " <> show e)+ -- | Force evaluation of a raw append shape result or fail the benchmark. forceRawAppend :: Either Pool.UsageError (Maybe RawAppendResult) -> IO () forceRawAppend (Right (Just (!streamId, !streamVersion, !globalPosition))) =@@ -657,6 +767,7 @@ main = do     -- Start ephemeral PostgreSQL once for all benchmarks     result <- Pg.withCached $ \db -> do+        migrateTestDatabase (Pg.connectionString db)         let settings = defaultConnectionSettings (Pg.connectionString db)         withStore settings $ \store -> do             -- Counter for unique stream names across benchmarks@@ -709,7 +820,12 @@                     r' <- runStoreIO store $ appendToStream sn NoStream [makeEvent "Init"]                     forceAppend r'                 )-                [StreamName "bench-multi-a", StreamName "bench-multi-b", StreamName "bench-multi-c"]+                ( [StreamName "bench-multi-a", StreamName "bench-multi-b", StreamName "bench-multi-c"]+                    <> map fst pipelinedMultiAppend4Streams+                    <> map fst pipelinedMultiAppend8Streams+                    <> map fst (concat pipelinedContentionCurrentGroups)+                    <> map fst (concat pipelinedContentionPipelinedGroups)+                )              -- Pre-create the hot stream targeted by the two-round-trip raw             -- shape variant. rawAppendUpdateExisting requires the row to@@ -757,6 +873,7 @@             concCounter <- newIORef (0 :: Int)             rawCounter <- newIORef (0 :: Int)             subCounter <- newIORef (0 :: Int)+            pipelinedContentionCounter <- newIORef (0 :: Int)              defaultMain                 [ bgroup@@ -833,6 +950,37 @@                             whnfIO $                                 runRawTwoRoundtripAppendExistingHotStreamTx store                         ]+                    ]+                , bgroup+                    "pipelined-multi-append"+                    [ bench "current shape (4 streams)" $+                        whnfIO $+                            runCurrentMultiAppend store pipelinedMultiAppend4Streams+                    , bench "pipelined (4 streams)" $+                        whnfIO $+                            runPipelinedMultiAppend store pipelinedMultiAppend4Streams+                    , bench "current shape (8 streams)" $+                        whnfIO $+                            runCurrentMultiAppend store pipelinedMultiAppend8Streams+                    , bench "pipelined (8 streams)" $+                        whnfIO $+                            runPipelinedMultiAppend store pipelinedMultiAppend8Streams+                    , bench "single-stream under 4 current multi-stream writers" $+                        whnfIO $+                            runSingleAppendUnderMultiWriters+                                store+                                pipelinedContentionCounter+                                runCurrentMultiAppend+                                "pipe-current"+                                pipelinedContentionCurrentGroups+                    , bench "single-stream under 4 pipelined multi-stream writers" $+                        whnfIO $+                            runSingleAppendUnderMultiWriters+                                store+                                pipelinedContentionCounter+                                runPipelinedMultiAppend+                                "pipe-pipelined"+                                pipelinedContentionPipelinedGroups                     ]                 , bgroup                     "read"

@@ -32,10 +32,10 @@ import Kiroku.Store.Subscription.Stream (subscriptionStream) import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..), SubscriptionHandleM (..)) import Shibuya.Adapter (Adapter (..))-import Shibuya.App (ProcessorId (..), SupervisionStrategy (..), mkProcessor, runApp, stopApp)+import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, stopApp) import Shibuya.Core.Ack (AckDecision (..)) import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Ingested (Ingested (..), Message) import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..)) import Shibuya.Telemetry.Effect (runTracingNoop) import Streamly.Data.Fold qualified as Fold@@ -199,14 +199,14 @@                     , shutdown = liftIO cancelAction                     } -        let handler :: (IOE :> es) => Ingested es RecordedEvent -> Eff es AckDecision+        let handler :: (IOE :> es) => Message es RecordedEvent -> Eff es AckDecision             handler _ingested = do                 liftIO $ atomically $ do                     c <- readTVar countVar                     writeTVar countVar (c + 1)                 pure AckOk -        res <- runApp IgnoreFailures 100 [(ProcessorId "bench", mkProcessor adapter handler)]+        res <- runApp defaultAppConfig [(ProcessorId "bench", mkProcessor adapter handler)]         case res of             Left err -> liftIO $ error ("runApp failed: " <> show err)             Right appHandle -> do@@ -228,6 +228,7 @@                 , partition = Nothing                 , enqueuedAt = Just (event ^. #createdAt)                 , traceContext = Nothing+                , headers = Nothing                 , attempt = Nothing                 , attributes = HashMap.empty                 , payload = event

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.2.0.0+version:         0.3.0.0 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -70,6 +70,7 @@     , hasql-transaction     >=1.1  && <1.3     , lens                  >=5.2  && <5.4     , mmzk-typeid           >=0.6  && <0.8+    , mtl                   >=2.3  && <2.4     , stm                   >=2.5  && <2.6     , streamly-core         >=0.3  && <0.4     , text                  >=2.0  && <2.2@@ -87,6 +88,7 @@   hs-source-dirs: test   other-modules:     Test.CatchupDbErrorNoPrematureSwitch+    Test.Category     Test.CategoryIdleNoSpin     Test.Causation     Test.Concurrency@@ -97,9 +99,14 @@     Test.FailureInjection     Test.Helpers     Test.InterpreterHooks+    Test.NotifyGuard     Test.Properties+    Test.PublisherCallbackResilience+    Test.PublisherIdleAdvance     Test.PublisherRestartNoRebroadcast     Test.ReadStream+    Test.StartupFailureSurfacing+    Test.StreamBridgeTermination     Test.StreamNameLookup     Test.SubscriptionPauseResume     Test.SubscriptionReconnect@@ -107,12 +114,14 @@     Test.SubscriptionRetryDeadLetter     Test.SubscriptionState     Test.Transaction+    Test.TruncateBefore    ghc-options:    -threaded -rtsopts -with-rtsopts=-N   build-depends:     , aeson                 >=2.1  && <2.3     , async                 >=2.2  && <2.3     , base                  >=4.18 && <5+    , bytestring            >=0.11 && <0.13     , containers            >=0.6  && <0.8     , contravariant-extras  >=0.3     , directory@@ -120,6 +129,7 @@     , ephemeral-pg          >=0.2  && <0.3     , generic-lens          >=2.2  && <2.4     , hasql                 >=1.10 && <1.11+    , hasql-notifications   >=0.2  && <0.3     , hasql-pool            >=1.2  && <1.5     , hasql-transaction     >=1.1  && <1.3     , hedgehog              >=1.4  && <1.8@@ -128,6 +138,7 @@     , kiroku-store     , kiroku-test-support     , lens                  >=5.2  && <5.4+    , mmzk-typeid           >=0.6  && <0.8     , stm                   >=2.5  && <2.6     , streamly-core         >=0.3  && <0.4     , text                  >=2.0  && <2.2@@ -142,23 +153,24 @@   hs-source-dirs: bench   ghc-options:    -threaded -rtsopts "-with-rtsopts=-N -A32m"   build-depends:-    , aeson              >=2.1  && <2.3-    , async              >=2.2  && <2.3-    , base               >=4.18 && <5-    , deepseq            >=1.4-    , ephemeral-pg       >=0.2  && <0.3-    , generic-lens       >=2.2  && <2.4-    , hasql              >=1.10 && <1.11-    , hasql-pool         >=1.2  && <1.5-    , hasql-transaction  >=1.1  && <1.3+    , aeson                >=2.1  && <2.3+    , async                >=2.2  && <2.3+    , base                 >=4.18 && <5+    , deepseq              >=1.4+    , ephemeral-pg         >=0.2  && <0.3+    , generic-lens         >=2.2  && <2.4+    , hasql                >=1.10 && <1.11+    , hasql-pool           >=1.2  && <1.5+    , hasql-transaction    >=1.1  && <1.3     , kiroku-store-    , lens               >=5.2  && <5.4-    , mmzk-typeid        >=0.6-    , tasty-bench        >=0.4-    , text               >=2.0  && <2.2-    , time               >=1.12 && <1.15-    , uuid               >=1.3  && <1.4-    , vector             >=0.13 && <0.14+    , kiroku-test-support+    , lens                 >=5.2  && <5.4+    , mmzk-typeid          >=0.6+    , tasty-bench          >=0.4+    , text                 >=2.0  && <2.2+    , time                 >=1.12 && <1.15+    , uuid                 >=1.3  && <1.4+    , vector               >=0.13 && <0.14  benchmark kiroku-shibuya-overhead   import:         common@@ -176,7 +188,7 @@     , generic-lens          >=2.2  && <2.4     , kiroku-store     , lens                  >=5.2  && <5.4-    , shibuya-core          >=0.6  && <0.7+    , shibuya-core          >=0.8  && <0.9     , stm                   >=2.5  && <2.6     , streamly              >=0.11     , streamly-core         >=0.3  && <0.4

@@ -58,14 +58,13 @@   existing stream (including soft-deleted). * 'Kiroku.Store.Error.DuplicateEvent' — caller-supplied 'eventId'   collides with an existing event.+* 'Kiroku.Store.Error.EmptyAppendBatch' — the supplied event list is+  empty.  Empty event lists: passing @[]@ is a programming mistake. The-underlying CTE returns 0 rows, which the interpreter maps to a-constructor that depends on the supplied 'ExpectedVersion'-('Kiroku.Store.Error.WrongExpectedVersion' for 'ExactVersion',-'Kiroku.Store.Error.StreamNotFound' for 'StreamExists', and so on).-None of these is the caller's intent; reject the call yourself before-invoking the API.+interpreter rejects the call with 'Kiroku.Store.Error.EmptyAppendBatch'+before any database work. This avoids taking the global @$all@ row lock,+firing notifications, or creating an empty stream under 'NoStream'. -} appendToStream ::     (HasCallStack, Store :> es) =>@@ -101,7 +100,9 @@ The returned list mirrors the input order. Each 'AppendResult' carries the corresponding stream's final state. -Empty input @[]@ is a no-op programming mistake; do not call.+Empty operation input @[]@ returns @[]@ without a database round trip.+Any per-stream empty event list is rejected with+'Kiroku.Store.Error.EmptyAppendBatch' before the transaction opens. -} appendMultiStream ::     (HasCallStack, Store :> es) =>

@@ -20,6 +20,7 @@ ) where  import Control.Lens ((^.))+import Control.Monad.Except qualified as Except import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value) import Data.Foldable (for_)@@ -40,14 +41,18 @@ import Effectful.Dispatch.Dynamic (interpret_) import Effectful.Error.Static (Error, runErrorNoCallStack, throwError) import GHC.Generics (Generic)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pipeline qualified as Pipeline import Hasql.Pool (Pool) import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session+import Hasql.Statement (Statement, unpreparable) import Hasql.Transaction qualified as Tx 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, mapUsageError)+import Kiroku.Store.Error (StoreError (..), attributeMultiStreamError, emptyResultError, isTransientSerializationError, mapLinkUsageError, mapTransactionUsageError, mapUsageError, validateStreamName) import Kiroku.Store.Observability (KirokuEvent (..)) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (decodeEvents, enrichEvents)@@ -74,6 +79,13 @@     Surfaced as 'Kiroku.Store.Read.lookupStreamId'.     -}     LookupStreamId :: StreamName -> Store m (Maybe StreamId)+    {- | Check whether an event id exists in a live stream without materializing+    stream events. Soft-deleted streams behave as nonexistent, mirroring+    'Kiroku.Store.Read.readStreamForward'.++    Surfaced as 'Kiroku.Store.Read.eventExistsInStream'.+    -}+    EventExistsInStream :: StreamName -> EventId -> Store m Bool     {- | Resolve a batch of surrogate 'StreamId's to their 'StreamName's in a     single round trip. The result 'Map' contains an entry only for ids that     name an existing stream (live or soft-deleted); hard-deleted or unknown ids@@ -103,6 +115,14 @@     SoftDeleteStream :: StreamName -> Store m (Maybe StreamId)     HardDeleteStream :: StreamName -> Store m (Maybe StreamId)     UndeleteStream :: StreamName -> Store m (Maybe StreamId)+    {- | Set a stream's logical truncate-before marker. Per-stream ordered+    reads then return only events whose version is >= the marker; the @$all@+    global log, category reads, and subscriptions are unaffected. Reversible.++    Surfaced as 'Kiroku.Store.Lifecycle.setStreamTruncateBefore' /+    'Kiroku.Store.Lifecycle.clearStreamTruncateBefore'.+    -}+    SetStreamTruncateBefore :: StreamName -> StreamVersion -> Store m (Maybe StreamId)     {- | Run an arbitrary @hasql-transaction@ value in a 'BEGIN'/'COMMIT'     block on a single pool connection. Escape hatch from the abstract     'Store' effect into the underlying SQL world; mock interpreters are@@ -130,20 +150,34 @@     Eff es a runStorePool store = interpret_ $ \case     AppendToStream (StreamName name) expected events -> do-        rejectReservedApplicationStream name+        rejectInvalidApplicationStream name+        case events of+            [] -> throwError (EmptyAppendBatch (StreamName name))+            _ -> pure ()         events' <- liftIO $ enrichEvents (store ^. #storeSettings) events         now <- liftIO getCurrentTime         prepared <- prepareEvents events'         let params = buildAppendParams name now prepared-        result <- liftIO $ Pool.use (store ^. #pool) $ case expected of-            ExactVersion (StreamVersion v) ->-                Session.statement (params, v) SQL.appendExpectedVersion-            StreamExists ->-                Session.statement params SQL.appendStreamExists-            NoStream ->-                Session.statement params SQL.appendNoStream-            AnyVersion ->-                Session.statement params SQL.appendAnyVersion+        let runOnce =+                Pool.use (store ^. #pool) $ case expected of+                    ExactVersion (StreamVersion v) ->+                        Session.statement (params, v) SQL.appendExpectedVersion+                    StreamExists ->+                        Session.statement params SQL.appendStreamExists+                    NoStream ->+                        Session.statement params SQL.appendNoStream+                    AnyVersion ->+                        Session.statement params SQL.appendAnyVersion+        firstAttempt <- liftIO runOnce+        result <- case firstAttempt of+            -- Match hasql-transaction's retryable SQLSTATE set, but only once:+            -- PostgreSQL rolls back the victim transaction and event ids were+            -- prepared before the first attempt, so this retry is idempotent.+            Left usageErr+                | isTransientSerializationError usageErr ->+                    liftIO runOnce+            _ ->+                pure firstAttempt         case result of             Left usageErr ->                 throwError (mapUsageError name expected usageErr)@@ -157,9 +191,10 @@                 Session.statement (name, startVer, limit) SQL.readStreamForwardStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs     ReadStreamBackward (StreamName name) (StreamVersion startVer) limit -> do+        let cursor = if startVer == 0 then maxBound else startVer         evs <-             usePool (store ^. #pool) $-                Session.statement (name, startVer, limit) SQL.readStreamBackwardStmt+                Session.statement (name, cursor, limit) SQL.readStreamBackwardStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs     ReadAllForward (GlobalPosition startPos) limit -> do         evs <-@@ -167,9 +202,10 @@                 Session.statement (startPos, limit) SQL.readAllForwardStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs     ReadAllBackward (GlobalPosition startPos) limit -> do+        let cursor = if startPos == 0 then maxBound else startPos         evs <-             usePool (store ^. #pool) $-                Session.statement (startPos, limit) SQL.readAllBackwardStmt+                Session.statement (cursor, limit) SQL.readAllBackwardStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs     GetStream (StreamName name) ->         usePool (store ^. #pool) $@@ -178,6 +214,11 @@         fmap (fmap StreamId) $             usePool (store ^. #pool) $                 Session.statement name SQL.findStreamIdStmt+    EventExistsInStream (StreamName name) (EventId eid) ->+        usePool (store ^. #pool) $+            Session.statement (name, eid) SQL.eventExistsInStreamStmt+    LookupStreamNames [] ->+        pure Map.empty     LookupStreamNames sids ->         fmap             (Map.fromList . map (\(s, nm) -> (StreamId s, StreamName nm)) . V.toList)@@ -185,23 +226,30 @@                 Session.statement [s | StreamId s <- sids] SQL.lookupStreamNamesStmt             )     LinkToStream (StreamName name) eventIds -> do-        rejectReservedApplicationStream name+        rejectInvalidApplicationStream name         let uuids = V.fromList [uid | EventId uid <- eventIds]         result <--            usePool (store ^. #pool) $-                Session.statement (uuids, name) SQL.linkToStreamStmt+            liftIO $+                Pool.use (store ^. #pool) $+                    Session.statement (uuids, name) SQL.linkToStreamStmt         case result of-            Nothing -> throwError (StreamNotFound (StreamName name))-            Just r -> pure r+            Left usageErr -> throwError (mapLinkUsageError (StreamName name) usageErr)+            Right Nothing -> throwError (StreamNotFound (StreamName name))+            Right (Just r) -> pure r     ReadCategoryForward (CategoryName cat) (GlobalPosition startPos) limit -> do         evs <-             usePool (store ^. #pool) $                 Session.statement (startPos, cat, limit) SQL.readCategoryForwardStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs+    AppendMultiStream [] ->+        pure []     AppendMultiStream ops -> do-        case find (\(StreamName name, _, _) -> isReservedApplicationStream name) ops of-            Just (sn, _, _) -> throwError (ReservedStreamName sn)+        case firstStreamNameError ops of+            Just err -> throwError err             Nothing -> pure ()+        case find (\(_, _, evts) -> null evts) ops of+            Just (sn, _, _) -> throwError (EmptyAppendBatch sn)+            Nothing -> pure ()         now <- liftIO getCurrentTime         -- Prepare all events for all streams         preparedOps <-@@ -213,27 +261,20 @@                 )                 ops         let names = V.fromList [n | (StreamName n, _, _) <- ops]-        let txn = do-                -- Pre-lock the user-named streams in deterministic (stream_id)-                -- order to avoid row-lock deadlocks between concurrent-                -- multi-stream txns touching overlapping streams in different-                -- user orders. See EP-1 F4.-                Tx.statement names SQL.lockStreamsForMultiStmt-                results <--                    mapM-                        ( \(StreamName name, expected, prepared) -> do-                            let params = buildAppendParams name now prepared-                            appendDispatchTx expected params-                        )-                        preparedOps-                -- If any result is Nothing (version conflict), condemn the transaction-                case any isNothing results of-                    True -> Tx.condemn >> pure results-                    False -> pure results-        result <--            liftIO $+        let runOnce =                 Pool.use (store ^. #pool) $-                    TxSessions.transaction TxSessions.ReadCommitted TxSessions.Write txn+                    runAppendMultiStreamPipeline names now preparedOps+        firstAttempt <- liftIO runOnce+        result <- case firstAttempt of+            -- Match the single-stream append retry and hasql-transaction's+            -- retryable SQLSTATE set. The failed transaction has rolled back,+            -- and event ids were prepared before the first attempt, so retrying+            -- once is idempotent.+            Left usageErr+                | isTransientSerializationError usageErr ->+                    liftIO runOnce+            _ ->+                pure firstAttempt         case result of             Left usageErr ->                 throwError (attributeMultiStreamError [(sn, ev) | (sn, ev, _) <- ops] usageErr)@@ -260,18 +301,22 @@                     Session.statement eid SQL.findCausationAncestorsStmt         liftIO $ decodeEvents (store ^. #storeSettings) evs     SoftDeleteStream (StreamName name) -> do-        rejectReservedApplicationStream name+        rejectInvalidApplicationStream name         usePool (store ^. #pool) $             Session.statement name SQL.softDeleteStreamStmt     HardDeleteStream (StreamName name) -> do-        rejectReservedApplicationStream name+        rejectInvalidApplicationStream name         let txn = do                 Tx.sql "SET LOCAL kiroku.enable_hard_deletes = 'on'"                 mSid <- Tx.statement name SQL.findStreamIdStmt                 case mSid of                     Nothing -> pure Nothing                     Just sid -> do-                        affected <- Tx.statement sid SQL.deleteStreamJunctionsStmt+                        originated <- Tx.statement sid SQL.deleteAllRowsForOriginStmt+                        Tx.statement originated SQL.deleteJunctionsByEventIdsStmt+                        linkedIn <- Tx.statement sid SQL.deleteStreamOwnJunctionsStmt+                        let affected = originated <> linkedIn+                        Tx.statement affected SQL.deleteDeadLettersForOrphanedEventsStmt                         Tx.statement affected SQL.deleteOrphanedEventsStmt                         Tx.statement sid SQL.deleteStreamRowStmt                         pure (Just (StreamId sid))@@ -287,9 +332,13 @@             Nothing -> pure ()         pure result     UndeleteStream (StreamName name) -> do-        rejectReservedApplicationStream name+        rejectInvalidApplicationStream name         usePool (store ^. #pool) $             Session.statement name SQL.undeleteStreamStmt+    SetStreamTruncateBefore (StreamName name) (StreamVersion v) -> do+        rejectInvalidApplicationStream name+        usePool (store ^. #pool) $+            Session.statement (name, v) SQL.setStreamTruncateBeforeStmt     RunTransaction tx ->         runTxOnPool (store ^. #pool) TxSessions.transaction tx     RunTransactionNoRetry tx ->@@ -344,21 +393,25 @@             Pool.use pool $                 entry TxSessions.ReadCommitted TxSessions.Write tx     case result of-        Left usageErr -> throwError (ConnectionError (T.pack (show usageErr)))+        Left usageErr -> throwError (mapTransactionUsageError usageErr)         Right a -> pure a --- | The seeded $all row is the global read stream, not an application stream.-isReservedApplicationStream :: Text -> Bool-isReservedApplicationStream = (== "$all")--rejectReservedApplicationStream ::+rejectInvalidApplicationStream ::     (Error StoreError :> es) =>     Text ->     Eff es ()-rejectReservedApplicationStream name-    | isReservedApplicationStream name = throwError (ReservedStreamName (StreamName name))-    | otherwise = pure ()+rejectInvalidApplicationStream name =+    either throwError pure (validateStreamName (StreamName name)) +firstStreamNameError :: [(StreamName, ExpectedVersion, [EventData])] -> Maybe StoreError+firstStreamNameError ops =+    case find (isLeft . validateStreamName . streamNameOf) ops of+        Just (sn, _, _) -> either Just (const Nothing) (validateStreamName sn)+        Nothing -> Nothing+  where+    streamNameOf (sn, _, _) = sn+    isLeft = either (const True) (const False)+ -- --------------------------------------------------------------------------- -- Internal helpers (moved from Append) -- ---------------------------------------------------------------------------@@ -412,6 +465,55 @@         , createdAts = V.fromList (replicate (length prepared) now)         , streamName = name         }++multiAppendBeginStmt :: Statement () ()+multiAppendBeginStmt = unpreparable "BEGIN" E.noParams D.noResult++multiAppendCommitStmt :: Statement () ()+multiAppendCommitStmt = unpreparable "COMMIT" E.noParams D.noResult++multiAppendRollbackStmt :: Statement () ()+multiAppendRollbackStmt = unpreparable "ROLLBACK" E.noParams D.noResult++runAppendMultiStreamPipeline ::+    Vector Text ->+    UTCTime ->+    [(StreamName, ExpectedVersion, [PreparedEvent])] ->+    Session.Session [Maybe AppendResult]+runAppendMultiStreamPipeline names now preparedOps =+    Except.catchError body $ \err -> do+        Session.statement () multiAppendRollbackStmt+        Except.throwError err+  where+    body = do+        results <-+            Session.pipeline $+                Pipeline.statement () multiAppendBeginStmt+                    -- Pre-lock the user-named streams in deterministic (stream_id)+                    -- order to avoid row-lock deadlocks between concurrent+                    -- multi-stream txns touching overlapping streams in different+                    -- user orders. See EP-1 F4.+                    *> Pipeline.statement names SQL.lockStreamsForMultiStmt+                    *> traverse appendPrepared preparedOps+        Session.statement () $+            if any isNothing results+                then multiAppendRollbackStmt+                else multiAppendCommitStmt+        pure results++    appendPrepared (StreamName name, expected, prepared) =+        appendDispatchPipeline expected (buildAppendParams name now prepared)++appendDispatchPipeline :: ExpectedVersion -> SQL.AppendParams -> Pipeline.Pipeline (Maybe AppendResult)+appendDispatchPipeline expected params = case expected of+    ExactVersion (StreamVersion v) ->+        Pipeline.statement (params, v) SQL.appendExpectedVersion+    StreamExists ->+        Pipeline.statement params SQL.appendStreamExists+    NoStream ->+        Pipeline.statement params SQL.appendNoStream+    AnyVersion ->+        Pipeline.statement params SQL.appendAnyVersion  {- | Dispatch the four 'SQL.append*' statements through 'Tx.statement', selecting the right one based on the supplied 'ExpectedVersion'.

@@ -1,21 +1,29 @@ module Kiroku.Store.Error (     StoreError (..),+    maxStreamNameBytes,+    validateStreamName,      -- * Append-precondition conflicts (Tx-flavored)     AppendConflict (..),     appendConflictToStoreError,     emptyResultConflict,     -- Internal helpers used by Effect module+    mapGenericUsageError,+    mapLinkUsageError,+    mapTransactionUsageError,     mapUsageError,     emptyResultError,     attributeMultiStreamError,+    isTransientSerializationError,     -- Internal pure helper exposed for unit testing     extractStreamNameFromDetail, ) where  import Control.Exception (Exception)+import Data.ByteString qualified as BS import Data.Text (Text) import Data.Text qualified as T+import Data.Text.Encoding qualified as TE import Data.UUID (UUID) import Data.UUID qualified as UUID import GHC.Generics (Generic)@@ -44,10 +52,21 @@ data StoreError     = {- | The actual stream version did not match the caller's       'ExactVersion' expectation. The third field is the actual-      version (or 'StreamVersion' 0 when the version could not be-      recovered, e.g., the row was concurrently soft-deleted).+      version placeholder. The append statement returns zero rows on a+      version mismatch and the store does not issue an extra read to+      recover the live version, so this field is 'StreamVersion' 0 on+      every empty-CTE rejection. Callers needing the live version must+      read the stream, for example with 'Kiroku.Store.Read.getStream'.       -}       WrongExpectedVersion !StreamName !ExpectedVersion !StreamVersion+    | {- | The caller supplied an empty event batch to an append surface.++      Appending zero events is always a programming mistake: before this+      guard existed, an empty batch silently took the global @$all@ row+      lock, fired NOTIFY triggers, and under 'NoStream' even created an+      empty stream. The interpreter rejects it before any pool work.+      -}+      EmptyAppendBatch !StreamName     | -- | The named stream does not exist (or has been soft-deleted).       StreamNotFound !StreamName     | {- | The named stream is reserved for store internals and cannot be@@ -56,6 +75,16 @@       @streams.stream_id = 0@ row.       -}       ReservedStreamName !StreamName+    | {- | The stream name exceeds 'maxStreamNameBytes' bytes of UTF-8.++      The append notification payload embeds the stream name, and+      PostgreSQL rejects @pg_notify@ payloads of 8,000 bytes or more.+      Enforcing a much smaller byte bound before any database work+      prevents an oversized stream name from failing late as an opaque+      trigger/server error. The second field is the offending UTF-8+      byte length.+      -}+      StreamNameTooLong !StreamName !Int     | {- | The named stream already exists. Returned for 'NoStream'       expectations against an existing stream and for 'linkToStream'       targets that already exist with conflicting state.@@ -70,6 +99,20 @@       offending id to the user should match on 'Just'.       -}       DuplicateEvent !(Maybe EventId)+    | {- | A 'Kiroku.Store.Link.linkToStream' call tried to link an+      event into a target stream that already contains it.++      Maps the @23505@ unique violation on @stream_events_pkey@, whose+      key is @(event_id, stream_id)@. Carries 'Just' the offending event+      id when PostgreSQL's detail string is parseable.+      -}+      EventAlreadyLinked !StreamName !(Maybe EventId)+    | {- | A 'Kiroku.Store.Link.linkToStream' call referenced an event id+      that does not exist. The link CTE surfaces this as a @23502@+      not-null violation on @stream_events.original_stream_id@ and the+      whole batch rolls back.+      -}+      LinkSourceEventMissing !StreamName     | {- | The connection pool timed out acquiring a connection. Almost       always retryable after a small backoff; sustained timeouts       indicate that the pool is undersized for the offered load or@@ -97,6 +140,19 @@     deriving stock (Eq, Show, Generic)     deriving anyclass (Exception) +-- | Maximum stream-name length in UTF-8 bytes.+maxStreamNameBytes :: Int+maxStreamNameBytes = 512++-- | Validate application stream names before any database work.+validateStreamName :: StreamName -> Either StoreError ()+validateStreamName sn@(StreamName name)+    | name == "$all" = Left (ReservedStreamName sn)+    | byteLength > maxStreamNameBytes = Left (StreamNameTooLong sn byteLength)+    | otherwise = Right ()+  where+    byteLength = BS.length (TE.encodeUtf8 name)+ {- | Map a hasql 'UsageError' to a 'StoreError'.  Pattern matches on the error hierarchy:@@ -113,7 +169,7 @@ and @ix_streams_stream_name@. If a future schema migration renames a constraint, the @23505@ branch falls through to the generic 'WrongExpectedVersion' mapping; keep the names stable in-@kiroku-store-migrations/sql-migrations@.+@kiroku-store-migrations/migrations@. -} mapUsageError :: Text -> ExpectedVersion -> UsageError -> StoreError mapUsageError streamName expected = \case@@ -124,6 +180,58 @@     AcquisitionTimeoutUsageError ->         PoolAcquisitionTimeout +{- | Map a hasql 'UsageError' raised inside an opaque transaction body.++The body has no stream context, so append duplicate violations are mapped+directly to 'DuplicateEvent' and all other failures fall back to the standard+append-shaped mapper with sentinel stream context.+-}+mapTransactionUsageError :: UsageError -> StoreError+mapTransactionUsageError usageErr =+    case extractServerError usageErr of+        Just (Errors.ServerError "23505" message detail _ _)+            | containsConstraint "events_pkey" message detail ->+                DuplicateEvent (EventId <$> (detail >>= extractUuidFromDetail))+            | containsConstraint "stream_events_pkey" message detail ->+                DuplicateEvent (EventId <$> (detail >>= extractFirstUuidFromCompositeDetail))+        _ ->+            mapUsageError "<transaction>" AnyVersion usageErr+  where+    containsConstraint name message detail =+        name `T.isInfixOf` message || maybe False (T.isInfixOf name) detail++-- | Generic, non-append-shaped mapping for hasql pool usage errors.+mapGenericUsageError :: UsageError -> StoreError+mapGenericUsageError = \case+    ConnectionUsageError connErr ->+        ConnectionLost (T.pack (show connErr))+    AcquisitionTimeoutUsageError ->+        PoolAcquisitionTimeout+    SessionUsageError sessionErr ->+        case extractServerError (SessionUsageError sessionErr) of+            Just (Errors.ServerError code message _ _ _) ->+                UnexpectedServerError code message+            Nothing ->+                ConnectionError ("Session error: " <> T.pack (show sessionErr))++-- | Map link-specific hasql failures to typed link errors.+mapLinkUsageError :: StreamName -> UsageError -> StoreError+mapLinkUsageError target usageErr =+    case extractServerError usageErr of+        Just (Errors.ServerError "23505" message detail _ _)+            | containsConstraint "stream_events_pkey" message detail ->+                EventAlreadyLinked target (extractCompositeEventId detail)+        Just (Errors.ServerError "23502" _ _ _ _) ->+            LinkSourceEventMissing target+        _ ->+            mapGenericUsageError usageErr+  where+    containsConstraint name message detail =+        name `T.isInfixOf` message || maybe False (T.isInfixOf name) detail++    extractCompositeEventId (Just d) = EventId <$> extractFirstUuidFromCompositeDetail d+    extractCompositeEventId Nothing = Nothing+ mapSessionError :: Text -> ExpectedVersion -> Errors.SessionError -> StoreError mapSessionError streamName expected = \case     Errors.StatementSessionError _ _ _ _ _ stmtErr ->@@ -192,11 +300,14 @@ -} data AppendConflict     = {- | Mirror of 'WrongExpectedVersion'. The third field is the-      actual stream version, or @StreamVersion 0@ when it could not-      be recovered (e.g., the CTE returned no rows because the-      target was concurrently soft-deleted).+      stream-version placeholder. The append statement returns zero+      rows on a version mismatch and the store does not issue an extra+      read to recover the live version, so this field is+      @'StreamVersion' 0@ on every empty-CTE rejection.       -}       WrongExpectedVersionConflict !StreamName !ExpectedVersion !StreamVersion+    | -- | Mirror of 'EmptyAppendBatch'.+      EmptyAppendBatchConflict !StreamName     | -- | Mirror of 'StreamNotFound'.       StreamNotFoundConflict !StreamName     | -- | Mirror of 'StreamAlreadyExists'.@@ -210,6 +321,7 @@ appendConflictToStoreError :: AppendConflict -> StoreError appendConflictToStoreError = \case     WrongExpectedVersionConflict sn ev sv -> WrongExpectedVersion sn ev sv+    EmptyAppendBatchConflict sn -> EmptyAppendBatch sn     StreamNotFoundConflict sn -> StreamNotFound sn     StreamAlreadyExistsConflict sn -> StreamAlreadyExists sn @@ -268,6 +380,19 @@                  in UUID.fromText uuidText         _ -> Nothing +{- | Extract the first UUID from a PostgreSQL composite-key detail string like:+"Key (event_id, stream_id)=(01234567-89ab-7def-8012-34567890abcd, 42) already exists."+-}+extractFirstUuidFromCompositeDetail :: Text -> Maybe UUID+extractFirstUuidFromCompositeDetail detail =+    case T.breakOn "=(" detail of+        (_, rest)+            | not (T.null rest) ->+                let afterParen = T.drop 2 rest -- skip "=("+                    uuidText = T.strip (T.takeWhile (\c -> c /= ',' && c /= ')') afterParen)+                 in UUID.fromText uuidText+        _ -> Nothing+ {- | Extract a stream name from a PostgreSQL unique-violation detail string like @"Key (stream_name)=(orders-1) already exists."@. @@ -345,3 +470,12 @@             Errors.ServerStatementError serverErr -> Just serverErr             _ -> Nothing     _ -> Nothing++-- | True for PostgreSQL transient transaction aborts retried by hasql-transaction.+isTransientSerializationError :: UsageError -> Bool+isTransientSerializationError usageErr =+    case extractServerError usageErr of+        Just (Errors.ServerError code _ _ _ _) ->+            code == "40001" || code == "40P01"+        Nothing ->+            False

@@ -2,6 +2,8 @@     softDeleteStream,     hardDeleteStream,     undeleteStream,+    setStreamTruncateBefore,+    clearStreamTruncateBefore, ) where  import Effectful (Eff, (:>))@@ -46,7 +48,7 @@ transaction. Direct @DELETE@ or @TRUNCATE@ against the underlying tables without setting that GUC raises an exception via the @protect_deletion@ and @protect_truncation@ triggers in-@kiroku-store-migrations/sql-migrations@.+@kiroku-store-migrations/migrations@.  /The GUC is an advisory protection, not a security boundary./ Any PostgreSQL session with @DELETE@ privilege on @events@,@@ -108,3 +110,42 @@     StreamName ->     Eff es (Maybe StreamId) undeleteStream name = send (UndeleteStream name)++{- | Set the logical truncate-before marker for a stream. After this call,+'Kiroku.Store.Read.readStreamForward' and 'readStreamBackward' return only the+events whose per-stream version is >= @before@; earlier events are hidden from+ordered stream reads but are NOT deleted.++This is the close-the-book / snapshot-and-compact primitive: append a snapshot+event (it lands at version @V@), then call @setStreamTruncateBefore name V@ so+rehydration starts from the snapshot.++The marker does NOT affect the @$all@ global log, 'Kiroku.Store.Read.readCategory',+subscriptions, or existence probes — the global history stays complete, so+projections (including ones not yet written) can still be built from it. The+operation is fully reversible: lower the marker or call+'clearStreamTruncateBefore' to re-expose hidden events.++Returns @Just streamId@ on success, @Nothing@ if the stream does not exist or is+soft-deleted. The reserved stream @$all@ is rejected with+'Kiroku.Store.Error.ReservedStreamName'. Per-stream versions are 1-based, so a+@before@ of 0 or 1 keeps the whole stream. Idempotent: setting the same value+again returns the same result and changes nothing.+-}+setStreamTruncateBefore ::+    (HasCallStack, Store :> es) =>+    StreamName ->+    StreamVersion ->+    Eff es (Maybe StreamId)+setStreamTruncateBefore name before = send (SetStreamTruncateBefore name before)++{- | Clear a stream's truncate-before marker, re-exposing the full history to+ordered stream reads. Equivalent to @setStreamTruncateBefore name 0@. Returns+@Just streamId@ on success, @Nothing@ for a missing or soft-deleted stream.+Rejects @$all@ with 'Kiroku.Store.Error.ReservedStreamName'.+-}+clearStreamTruncateBefore ::+    (HasCallStack, Store :> es) =>+    StreamName ->+    Eff es (Maybe StreamId)+clearStreamTruncateBefore name = setStreamTruncateBefore name (StreamVersion 0)

@@ -10,6 +10,15 @@  {- | Link existing events into a target stream. +__Provisional API.__ No known consumer uses this function (audited+2026-06-11: zero usage in keiro, the only downstream framework). It is the+only public feature that requires the @stream_events@ junction-table layout,+which a future global-position migration may replace with a single-table event+layout (global position as a column on @events@); in that case this function+will be removed or redesigned (e.g. rehomed to a dedicated @stream_links@ side+table). If you have a real use case, surface it before depending on this. See+docs/architecture/global-position-migration-path.md.+ Creates the target stream if it does not exist; otherwise appends links to the existing target. Linking does not duplicate the event payload — the @events@ row is shared, and only a new junction row in@@ -23,7 +32,8 @@ * Every supplied 'Kiroku.Store.Types.EventId' must reference an event   that currently exists. Linking an unknown id (a typo, or an id that   was hard-deleted) rejects the entire batch atomically — the target-  stream is left in its pre-call state. (See EP-1 F3.)+  stream is left in its pre-call state — with+  'Kiroku.Store.Error.LinkSourceEventMissing'. (See EP-1 F3.) * The target must not be soft-deleted. Linking to a soft-deleted target   fails with 'Kiroku.Store.Error.StreamNotFound'. (See EP-1 F5.) * The target must not be @$all@. @$all@ is the global read stream, and@@ -31,8 +41,7 @@   the interpreter rejects this with   'Kiroku.Store.Error.ReservedStreamName'. * Linking the same event into the same target stream twice fails with-  a primary-key violation on the junction's @(stream_id, event_id)@-  uniqueness.+  'Kiroku.Store.Error.EventAlreadyLinked'.  Returns the target's 'Kiroku.Store.Types.LinkResult' — its id and the position of the /last/ linked event in the target.

@@ -9,12 +9,11 @@ import Control.Concurrent.Async (Async) import Control.Concurrent.Async qualified as Async import Control.Concurrent.STM (TChan, TVar, atomically, modifyTVar', newBroadcastTChanIO, newTVarIO, readTVarIO, writeTChan, writeTVar)-import Control.Exception (Exception, SomeException, asyncExceptionFromException, bracketOnError, catch, throwIO, toException)+import Control.Exception (Exception, SomeAsyncException, SomeException, asyncExceptionFromException, bracketOnError, catch, throwIO, toException) import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString (ByteString) import Data.ByteString.Char8 qualified as BC-import Data.Foldable (for_) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Text (Text)@@ -30,7 +29,8 @@ import Hasql.Notifications qualified as Notifications import Hasql.Session qualified as Session import Hasql.Statement (Statement, unpreparable)-import Kiroku.Store.Observability (KirokuEvent (..))+import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop)+import Kiroku.Store.Types (CategoryName (..), StreamName (..), categoryName)  {- | A Notifier manages a dedicated PostgreSQL connection for LISTEN/NOTIFY. It writes a @()@ tick to a broadcast 'TChan' on every notification,@@ -61,15 +61,15 @@     -}     } -{- | Raised by 'startNotifier' when the initial dedicated @LISTEN@-connection cannot be acquired. Carries @hasql@'s underlying-'Connection.ConnectionError' (often a TCP-level failure or an-authentication error) for diagnostics.+{- | Raised by 'startNotifier' when the initial dedicated @LISTEN@ setup fails. -Replaces the prior @IOException@-via-@fail@ shape so callers can pattern-match on a typed startup exception.+Connection-acquire failures carry @hasql@'s underlying+'Connection.ConnectionError'. LISTEN failures carry the exception thrown after+the connection was acquired; that connection has already been released. -}-newtype NotifierStartError = NotifierStartError ConnectionError+data NotifierStartError+    = NotifierConnectError !ConnectionError+    | NotifierListenError !SomeException     deriving stock (Show)     deriving anyclass (Exception) @@ -108,7 +108,7 @@  On 'Async.AsyncCancelled' (from cancellation), exits cleanly. -Initial-acquire failure raises 'NotifierStartError'.+Initial acquire or LISTEN failure raises 'NotifierStartError'. -} startNotifier ::     (MonadIO m) =>@@ -122,18 +122,22 @@ startNotifier connString schema mHandler = liftIO $ do     chan <- newBroadcastTChanIO     catGenVar <- newTVarIO Map.empty-    conn <- acquireOrThrow connString     let channel = toPgIdentifier (schema <> ".events")-    Notifications.listen conn channel-    connRef <- newTVarIO conn-    thread <- Async.async (listenerLoop chan catGenVar connRef channel connString mHandler)-    pure-        Notifier-            { tickChan = chan-            , listenerThread = thread-            , listenerConnRef = connRef-            , categoryGenerations = catGenVar-            }+    bracketOnError (acquireOrThrow connString) Connection.release $ \conn -> do+        Notifications.listen conn channel+            `catch` \(e :: SomeException) ->+                case asyncExceptionFromException e of+                    Just (ae :: SomeAsyncException) -> throwIO ae+                    Nothing -> throwIO (NotifierListenError e)+        connRef <- newTVarIO conn+        thread <- Async.async (listenerLoop chan catGenVar connRef channel connString mHandler)+        pure+            Notifier+                { tickChan = chan+                , listenerThread = thread+                , listenerConnRef = connRef+                , categoryGenerations = catGenVar+                }  {- | Stop the Notifier. Cancels the listener thread, waits for it to finish, and releases whichever connection the loop is currently holding.@@ -210,7 +214,7 @@                 emit KirokuEventNotifierReconnected                 go -    emit evt = for_ mHandler ($ evt)+    emit = emitOrDrop mHandler  -- The 'waitForNotifications' callback. Wakes the publisher (the bare @()@ tick, -- preserving the existing AllStreams broadcast path) AND bumps the originating@@ -228,14 +232,15 @@ -- payload is @stream_name,stream_id,stream_version@; @stream_id@ and -- @stream_version@ are comma-free integers, so the @stream_name@ is everything -- except the last two comma-separated fields (rejoined, since a stream name may--- itself contain commas). The category is then the prefix before the first @-@,--- matching the @streams.category GENERATED ALWAYS AS split_part(stream_name,'-',1)@--- column. A category never contains @-@, so @takeWhile (/= '-')@ is exact.+-- itself contain commas). The category is then derived by 'categoryName' — the+-- single Haskell definition of the "prefix before the first @-@" rule, matching+-- the @streams.category GENERATED ALWAYS AS split_part(stream_name,'-',1)@ column. categoryFromPayload :: ByteString -> Text categoryFromPayload payload =     let fields = BC.split ',' payload-        streamName = BC.intercalate "," (take (max 0 (length fields - 2)) fields)-     in decodeUtf8Lenient (BC.takeWhile (/= '-') streamName)+        streamNameBytes = BC.intercalate "," (take (max 0 (length fields - 2)) fields)+        CategoryName category = categoryName (StreamName (decodeUtf8Lenient streamNameBytes))+     in category  -- A synthetic exception used when 'waitForNotifications' returns without -- raising (hasql-notifications turned a connection error into a Left@@ -246,14 +251,13 @@     deriving anyclass (Exception)  -- Internal: acquire a connection, set its application_name, or throw on failure.--- Throws 'NotifierStartError' (which derives 'Exception') instead of the--- prior @fail@-based 'IOException', so callers can match on a typed startup--- exception.+-- Throws 'NotifierStartError' (which derives 'Exception') for connection+-- acquisition failures, so callers can match on a typed startup exception. acquireOrThrow :: Text -> IO Connection acquireOrThrow connStr = do     result <- Connection.acquire (Conn.connectionString connStr)     case result of-        Left err -> throwIO (NotifierStartError err)+        Left err -> throwIO (NotifierConnectError err)         Right conn -> do             -- Tag the connection so operators can identify the listener in             -- pg_stat_activity. Failures here are non-fatal — fall back to the

@@ -41,9 +41,11 @@     SubscriptionStopReason (..),     SubscriptionGroupContext (..),     DeadLetterReason (..),+    emitOrDrop, ) where -import Control.Exception (SomeException)+import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)+import Data.Foldable (for_) import Data.Int (Int32) import Hasql.Pool (UsageError) import Kiroku.Store.Subscription.Fsm (DeadLetterReason (..), SubscriptionStopReason (..))@@ -77,20 +79,27 @@       the application pool) or a persistent server error.       -}       KirokuEventPublisherPoolError !UsageError+    | {- | The publisher loop's broadcast iteration threw a synchronous+      exception that was not a 'UsageError' — typically a user-supplied+      'Kiroku.Store.Settings.decodeHook' or observability handler throwing.+      The publisher skipped this tick and will retry on the next notification+      or the 30-second safety poll. Sustained emissions mean a deterministically+      failing callback is stalling live broadcast until it is fixed.+      -}+      KirokuEventPublisherLoopError !SomeException     | {- | A subscription's worker thread encountered a 'UsageError' in-      the database phase identified by 'SubscriptionDbPhase'. The-      worker may continue with a documented fallback for checkpoint-      load/save phases, while fetch-batch errors are retried at the same-      cursor. The event is the operator's structured signal that this-      happened. The trailing 'SubscriptionGroupContext' identifies which-      consumer-group member (if any) emitted it.+      the database phase identified by 'SubscriptionDbPhase'. Checkpoint-load+      errors fail startup loudly, fetch-batch errors are retried at the same+      cursor, and checkpoint-save errors are reported while the worker continues.+      The event is the operator's structured signal that this happened. The+      trailing 'SubscriptionGroupContext' identifies which consumer-group member+      (if any) emitted it.       -}       KirokuEventSubscriptionDbError !SubscriptionName !SubscriptionDbPhase !UsageError !SubscriptionGroupContext     | {- | A subscription's worker thread has just started; the worker-      will begin from the recorded 'GlobalPosition' (zero if no-      checkpoint exists or 'KirokuEventSubscriptionDbError' fired in-      the @LoadCheckpoint@ phase). The trailing 'SubscriptionGroupContext'-      identifies which consumer-group member (if any) started.+      will begin from the recorded 'GlobalPosition' (zero only when no+      checkpoint exists). The trailing 'SubscriptionGroupContext' identifies+      which consumer-group member (if any) started.       -}       KirokuEventSubscriptionStarted !SubscriptionName !GlobalPosition !SubscriptionGroupContext     | {- | The subscription has reached the EventPublisher's@@ -189,13 +198,26 @@       KirokuEventHardDeleteIssued !StreamName !StreamId     deriving stock (Show) +{- | Invoke the optional observability handler, dropping any synchronous exception+it throws. Asynchronous exceptions, including thread cancellation, are rethrown.++Store-internal threads must not die because a metrics or logging callback threw;+the callback contract is that it should be fast and non-throwing.+-}+emitOrDrop :: Maybe (KirokuEvent -> IO ()) -> KirokuEvent -> IO ()+emitOrDrop mHandler evt =+    for_ mHandler $ \handler ->+        handler evt `catch` \(e :: SomeException) ->+            case asyncExceptionFromException e of+                Just (ae :: SomeAsyncException) -> throwIO ae+                Nothing -> pure ()+ -- | Which database phase a 'KirokuEventSubscriptionDbError' fired in. data SubscriptionDbPhase     = {- | 'Kiroku.Store.Subscription.Worker' failed to read the saved-      checkpoint at subscription startup. The worker continues with-      'Kiroku.Store.Types.GlobalPosition' @0@; on a fresh subscription-      this is correct, on an existing subscription it silently-      re-processes events.+      checkpoint at subscription startup. The worker fails startup loudly+      instead of silently re-processing from 'Kiroku.Store.Types.GlobalPosition'+      @0@.       -}       LoadCheckpoint     | {- | The worker's catch-up or category-live database fetch

@@ -7,6 +7,7 @@     readCategory,     getStream,     lookupStreamId,+    eventExistsInStream,     lookupStreamName,     lookupStreamNames, ) where@@ -48,8 +49,10 @@ The streaming sibling of 'readStreamForward'. Identical SQL path and identical error semantics: this function dispatches 'readStreamForward' repeatedly with the supplied @pageSize@ as the per-call limit, advancing the exclusive-'StreamVersion' cursor across pages until the next call returns an empty-batch.+'StreamVersion' cursor across pages until a page comes back shorter than+@pageSize@. If the stream length is an exact multiple of @pageSize@, one+final empty-page probe is still required because the wrapper does not know+the total stream length up front.  The exclusive-cursor convention is preserved end-to-end: passing @'StreamVersion' 0@ reads from the first event in the stream. Empty and@@ -67,24 +70,31 @@     Int32 ->     Stream (Eff es) RecordedEvent readStreamForwardStream name startVer pageSize =-    Stream.concatMap (Stream.fromList . V.toList) pages+    Stream.concatMap fromVector pages   where-    pages = Stream.unfoldrM nextPage startVer-    nextPage cursor = do+    pages = Stream.unfoldrM nextPage (Just startVer)+    nextPage Nothing = pure Nothing+    nextPage (Just cursor) = do         events <- readStreamForward name cursor pageSize         if V.null events             then pure Nothing-            else-                let lastV = V.last events ^. #streamVersion-                 in pure (Just (events, lastV))+            else do+                let nextState+                        | V.length events < fromIntegral pageSize = Nothing+                        | otherwise = Just (V.last events ^. #streamVersion)+                pure (Just (events, nextState)) +    fromVector :: Vector a -> Stream (Eff es) a+    fromVector v = Stream.unfoldr (\i -> fmap (\x -> (x, i + 1)) (v V.!? i)) 0+ {- | Read events from a named stream in backward (descending version) order.  The cursor is exclusive: events with @streamVersion < startVer@ are returned (events older than the cursor). To read the entire stream from-the latest event backward, pass @'StreamVersion' 0@ (the SQL treats it-as \"newer than any\"). Returns an empty vector for nonexistent or+the latest event backward, pass @'StreamVersion' 0@. The interpreter maps+0 to the maximum cursor value, so it never collides with a real stream+version; versions start at 1. Returns an empty vector for nonexistent or soft-deleted streams. -} readStreamBackward ::@@ -116,7 +126,8 @@ ('GlobalPosition'-descending) order.  Cursor exclusive. To start from the most recent event, pass-@'GlobalPosition' 0@ (treated as \"after everything\" by the SQL).+@'GlobalPosition' 0@. The interpreter maps 0 to the maximum cursor value,+so it never collides with a real global position; positions start at 1. -} readAllBackward ::     (HasCallStack, Store :> es) =>@@ -170,6 +181,20 @@     Eff es (Maybe StreamId) lookupStreamId name = send (LookupStreamId name) +{- | Return 'True' when the supplied event id is linked into the supplied live+stream, and 'False' when the event id is absent, belongs only to another stream,+or the stream is nonexistent or soft-deleted.++This is a point lookup for idempotency checks. Prefer it over scanning a stream+when the caller already has the event id it is looking for.+-}+eventExistsInStream ::+    (HasCallStack, Store :> es) =>+    StreamName ->+    EventId ->+    Eff es Bool+eventExistsInStream name eid = send (EventExistsInStream name eid)+ {- | Resolve a batch of surrogate 'StreamId's to their 'StreamName's in a single round trip, returning a 'Map' that omits any id which does not name an existing stream (hard-deleted or never created). Live and soft-deleted streams are@@ -188,8 +213,8 @@ -- names '!?' (event '^.' #originalStreamId) :: Maybe StreamName @ -Passing @[]@ returns an empty map without a database round trip's worth of work-(an empty @ANY(ARRAY[])@ matches nothing).+Passing @[]@ returns an empty map without any database round trip because+the interpreter short-circuits. -} lookupStreamNames ::     (HasCallStack, Store :> es) =>

@@ -18,6 +18,7 @@     readAllBackwardStmt,     readCategoryForwardStmt,     getStreamStmt,+    eventExistsInStreamStmt,     lookupStreamNamesStmt,     currentGlobalPositionStmt, @@ -33,10 +34,14 @@     -- * Lifecycle statements     softDeleteStreamStmt,     undeleteStreamStmt,+    setStreamTruncateBeforeStmt,      -- * Hard-delete statements (used in sequence inside one transaction)     findStreamIdStmt,-    deleteStreamJunctionsStmt,+    deleteAllRowsForOriginStmt,+    deleteJunctionsByEventIdsStmt,+    deleteStreamOwnJunctionsStmt,+    deleteDeadLettersForOrphanedEventsStmt,     deleteOrphanedEventsStmt,     deleteStreamRowStmt, @@ -386,7 +391,7 @@         <*> D.column (D.nullable D.uuid)         <*> D.column (D.nonNullable D.timestamptz) --- | Shared decoder for a StreamInfo row (5 columns).+-- | Shared decoder for a StreamInfo row (6 columns). streamInfoRow :: D.Row StreamInfo streamInfoRow =     StreamInfo@@ -395,6 +400,7 @@         <*> (StreamVersion <$> D.column (D.nonNullable D.int8))         <*> D.column (D.nonNullable D.timestamptz)         <*> D.column (D.nullable D.timestamptz)+        <*> (StreamVersion <$> D.column (D.nonNullable D.int8)) -- truncate_before  -- | Encoder for stream read params: (stream_name, start_version, limit). readStreamEncoder :: E.Params (Text, Int64, Int32)@@ -459,13 +465,34 @@         (E.param (E.nonNullable E.text))         (D.rowMaybe streamInfoRow) +-- | Point probe for whether an event exists in a live stream.+eventExistsInStreamStmt :: Statement (Text, UUID) Bool+eventExistsInStreamStmt =+    preparable+        """+        SELECT EXISTS (+          SELECT 1+          FROM stream_events se+          WHERE se.event_id = $2+            AND se.stream_id = (+              SELECT stream_id+              FROM streams+              WHERE stream_name = $1+                AND deleted_at IS NULL+            )+        )+        """+        (contrazip2 (E.param (E.nonNullable E.text)) (E.param (E.nonNullable E.uuid)))+        (D.singleRow (D.column (D.nonNullable D.bool)))+ -- --------------------------------------------------------------------------- -- Read SQL Templates -- ---------------------------------------------------------------------------  {- | Read from a named stream in forward order.-Resolves stream name to ID via subquery, joins stream_events with events,-filters on stream_version > start_version, orders ascending, limits.+Joins stream_events with events and streams, filters on stream_version >+start_version, hides the logically-truncated prefix (stream_version >=+streams.truncate_before), orders ascending, limits. For stream reads, global_position is set to 0 (not available without $all join). -} readStreamForwardSQL :: Text@@ -477,14 +504,21 @@            e.data, e.metadata, e.causation_id, e.correlation_id,            e.created_at     FROM stream_events se-    JOIN events e ON e.event_id = se.event_id-    WHERE se.stream_id = (SELECT stream_id FROM streams WHERE stream_name = $1 AND deleted_at IS NULL)+    JOIN events e  ON e.event_id  = se.event_id+    JOIN streams s ON s.stream_id = se.stream_id+    WHERE s.stream_name = $1+      AND s.deleted_at IS NULL       AND se.stream_version > $2+      AND se.stream_version >= s.truncate_before     ORDER BY se.stream_version ASC     LIMIT $3     """ --- | Read from a named stream in backward order.+{- | Read from a named stream in backward order.+The start version is an exclusive upper bound. The interpreter maps+caller cursor 0 to 'maxBound' so "from latest" stays a Haskell API+sentinel and the SQL keeps a plain range predicate.+-} readStreamBackwardSQL :: Text readStreamBackwardSQL =     """@@ -494,15 +528,21 @@            e.data, e.metadata, e.causation_id, e.correlation_id,            e.created_at     FROM stream_events se-    JOIN events e ON e.event_id = se.event_id-    WHERE se.stream_id = (SELECT stream_id FROM streams WHERE stream_name = $1 AND deleted_at IS NULL)-      AND se.stream_version > $2+    JOIN events e  ON e.event_id  = se.event_id+    JOIN streams s ON s.stream_id = se.stream_id+    WHERE s.stream_name = $1+      AND s.deleted_at IS NULL+      AND se.stream_version < $2+      AND se.stream_version >= s.truncate_before     ORDER BY se.stream_version DESC     LIMIT $3     """  {- | Read from the global $all stream in forward order. stream_id = 0 is the $all stream. stream_version on $all is the global position.+The global log deliberately ignores the per-stream @truncate_before@ marker:+logical truncation only hides events from ordered per-stream reads, never from+the @$all@ backbone that subscriptions and projections consume. -} readAllForwardSQL :: Text readAllForwardSQL =@@ -520,7 +560,11 @@     LIMIT $2     """ --- | Read from the global $all stream in backward order.+{- | Read from the global $all stream in backward order.+The start position is an exclusive upper bound. The interpreter maps+caller cursor 0 to 'maxBound' so "from latest" stays a Haskell API+sentinel and the SQL keeps a plain range predicate.+-} readAllBackwardSQL :: Text readAllBackwardSQL =     """@@ -532,7 +576,7 @@     FROM stream_events se     JOIN events e ON e.event_id = se.event_id     WHERE se.stream_id = 0-      AND se.stream_version > $1+      AND se.stream_version < $1     ORDER BY se.stream_version DESC     LIMIT $2     """@@ -541,7 +585,7 @@ getStreamSQL :: Text getStreamSQL =     """-    SELECT stream_id, stream_name, stream_version, created_at, deleted_at+    SELECT stream_id, stream_name, stream_version, created_at, deleted_at, truncate_before     FROM streams     WHERE stream_name = $1     """@@ -892,6 +936,16 @@         (E.param (E.nonNullable E.text))         (D.rowMaybe (StreamId <$> D.column (D.nonNullable D.int8))) +{- | Set a stream's logical truncate-before marker. Returns Nothing if the+stream does not exist or is soft-deleted.+-}+setStreamTruncateBeforeStmt :: Statement (Text, Int64) (Maybe StreamId)+setStreamTruncateBeforeStmt =+    preparable+        setStreamTruncateBeforeSQL+        (contrazip2 (E.param (E.nonNullable E.text)) (E.param (E.nonNullable E.int8)))+        (D.rowMaybe (StreamId <$> D.column (D.nonNullable D.int8)))+ {- | Look up a stream's id by name. Returns Nothing if the stream does not exist. Used as the first step in hard-delete; subsequent steps key off the id rather than the name so a single resolution is reused.@@ -920,43 +974,107 @@             )         ) -{- | Delete every junction row that references the given stream — both rows whose-@stream_id@ is the target (the stream's own and any links from elsewhere) and-rows whose @original_stream_id@ is the target ($all entries plus link rows in-other streams that referenced events originating from the deleted stream).+{- | Delete the @$all@ junction rows for events that originated in the target+stream, returning exactly those originated event ids. -Returns the distinct set of @event_id@s that lost at least one junction row.-The caller passes this set to @deleteOrphanedEventsStmt@ to remove event payloads-that no longer have any surviving junctions.+Served by @ix_stream_events_all_by_origin@, whose partial predicate is+@stream_id = 0@. Every event that originated in a stream has one @$all@ row, so+this indexed delete yields the complete originated-event set without scanning all+of @stream_events@. -The previous single-CTE implementation tried to inline the orphan-event delete,-but PostgreSQL §7.8.2 specifies that data-modifying CTEs run against the same-snapshot, so the @NOT EXISTS@ subquery on stream_events saw the pre-delete state-and never deleted any events. Splitting into two statements within the same-hasql-transaction lets each statement see the previous statement's effects.+The caller then passes the returned ids to 'deleteJunctionsByEventIdsStmt' to+remove the remaining home/link junction rows for those originated events. -}-deleteStreamJunctionsStmt :: Statement Int64 (Vector UUID)-deleteStreamJunctionsStmt =+deleteAllRowsForOriginStmt :: Statement Int64 (Vector UUID)+deleteAllRowsForOriginStmt =     preparable         """-        WITH deleted AS (-          DELETE FROM stream_events-          WHERE stream_id = $1-             OR original_stream_id = $1-          RETURNING event_id-        )-        SELECT DISTINCT event_id FROM deleted+        DELETE FROM stream_events+        WHERE original_stream_id = $1+          AND stream_id = 0+        RETURNING event_id         """         (E.param (E.nonNullable E.int8))         (D.rowVector (D.column (D.nonNullable D.uuid))) +{- | Delete all remaining junction rows for the given event ids.++Served by the @stream_events@ primary key @(event_id, stream_id)@. This removes+the originated events' home rows and any link rows in other streams after+'deleteAllRowsForOriginStmt' has removed their @$all@ rows.+-}+deleteJunctionsByEventIdsStmt :: Statement (Vector UUID) ()+deleteJunctionsByEventIdsStmt =+    preparable+        """+        DELETE FROM stream_events+        WHERE event_id = ANY($1::uuid[])+        """+        (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+        D.noResult++{- | Delete junction rows whose visible stream is the target stream, returning+the affected event ids.++Served by @ix_stream_events_stream_version@ (or the unique replacement+@ux_stream_events_stream_version@ after EP-5 M3) on @(stream_id,+stream_version)@. After 'deleteJunctionsByEventIdsStmt' this matches only+events that originated elsewhere and were linked into the deleted stream. Those+events normally survive; returning them lets the same orphan check make the+decision instead of relying on that assumption.+-}+deleteStreamOwnJunctionsStmt :: Statement Int64 (Vector UUID)+deleteStreamOwnJunctionsStmt =+    preparable+        """+        DELETE FROM stream_events+        WHERE stream_id = $1+        RETURNING event_id+        """+        (E.param (E.nonNullable E.int8))+        (D.rowVector (D.column (D.nonNullable D.uuid)))++{- | Delete dead-letter rows for candidate events that have become orphaned.++Served by @ix_dead_letters_event_id@. The @NOT EXISTS@ predicate mirrors+'deleteOrphanedEventsStmt', so dead letters are purged exactly for events whose+payload rows are about to be removed while dead letters for surviving linked-in+events remain.++Must be called after the junction deletes and before 'deleteOrphanedEventsStmt'+within the same transaction so the @NOT EXISTS@ subquery sees the post-delete+state of @stream_events@ and the @dead_letters.event_id@ foreign key remains+satisfied when orphan event payloads are deleted.+-}+deleteDeadLettersForOrphanedEventsStmt :: Statement (Vector UUID) ()+deleteDeadLettersForOrphanedEventsStmt =+    preparable+        """+        DELETE FROM dead_letters dl+        WHERE dl.event_id = ANY($1::uuid[])+          AND NOT EXISTS (+            SELECT 1 FROM stream_events se+            WHERE se.event_id = dl.event_id+          )+        """+        (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+        D.noResult+ {- | Delete event payloads from the @events@ table for the given event ids, but only those whose junction rows have all been removed. Events that still have any surviving @stream_events@ row are preserved (they remain visible from their other homes). -Must be called after @deleteStreamJunctionsStmt@ within the same transaction-so the @NOT EXISTS@ subquery sees the post-delete state of @stream_events@.+The first implementation tried to inline orphan-event deletion with a+data-modifying CTE, but PostgreSQL §7.8.2 specifies that data-modifying CTEs run+against the same snapshot, so the @NOT EXISTS@ subquery on stream_events saw the+pre-delete state and never deleted any events. Splitting into multiple+statements within the same hasql-transaction lets each statement see previous+statements' effects.++Must be called after 'deleteDeadLettersForOrphanedEventsStmt' within the same+transaction so the @NOT EXISTS@ subquery sees the post-delete state of+@stream_events@ and the @dead_letters.event_id@ foreign key is already clear. -} deleteOrphanedEventsStmt :: Statement (Vector UUID) () deleteOrphanedEventsStmt =@@ -1029,6 +1147,20 @@     SET deleted_at = NULL     WHERE stream_name = $1       AND deleted_at IS NOT NULL+    RETURNING stream_id+    """++{- | Set the logical truncate-before marker on a live stream. Soft-deleted+streams (deleted_at IS NOT NULL) match no row and return Nothing, matching+the other lifecycle statements.+-}+setStreamTruncateBeforeSQL :: Text+setStreamTruncateBeforeSQL =+    """+    UPDATE streams+    SET truncate_before = $2+    WHERE stream_name = $1+      AND deleted_at IS NULL     RETURNING stream_id     """ 

@@ -13,7 +13,7 @@  import Control.Concurrent.Async qualified as Async import Control.Concurrent.STM (atomically, modifyTVar', newTVarIO, readTVarIO)-import Control.Exception (bracket, finally, throwIO)+import Control.Exception (bracket, bracketOnError, finally, mask, throwIO) import Control.Lens ((^.)) import Control.Monad (when) import Control.Monad.IO.Class (MonadIO, liftIO)@@ -31,23 +31,24 @@ import Kiroku.Store.Subscription.EventPublisher qualified as Pub import Kiroku.Store.Subscription.Fsm (SubscriptionState (..), stateCursor, stateName) import Kiroku.Store.Subscription.Types-import Kiroku.Store.Subscription.Worker (configMember, runWorker)-import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Store.Subscription.Worker (LiveSource (..), configMember, runWorker)+import Kiroku.Store.Types (CategoryName (..), GlobalPosition (..))  {- | Start a subscription. Returns a handle for cancellation and waiting.  The subscription spawns a worker thread that:  1. Reads the checkpoint from the database (or starts from global position 0-   for a fresh subscription name).+   for a fresh subscription name). A database error while loading the checkpoint+   fails the worker loudly through 'wait'; it does not fall back to 0. 2. Catches up by querying the database directly until it reaches the    'Kiroku.Store.Subscription.EventPublisher.lastPublished' cursor. 3. Switches to live mode. For 'Kiroku.Store.Subscription.Types.AllStreams'-   subscriptions, the worker reads pre-broadcast events from the-   publisher's bounded per-subscriber queue. For-   'Kiroku.Store.Subscription.Types.Category' subscriptions, the worker-   bypasses the broadcast entirely and re-queries the database with the-   SQL category filter whenever 'lastPublished' advances.+   subscriptions outside a consumer group, the worker reads pre-broadcast+   events from the publisher's bounded per-subscriber queue. For+   'Kiroku.Store.Subscription.Types.Category' subscriptions and all+   consumer-group subscriptions, no publisher queue is created; the worker+   re-queries the database in live mode.  The returned 'SubscriptionHandle' carries an implicit lifecycle contract: the worker thread runs until 'cancel' is called or the handler returns 'Stop'.@@ -92,11 +93,17 @@   in flight at cancellation time will replay. * @Left e@ where @e@ is   'Kiroku.Store.Subscription.Types.SubscriptionOverflowed' — the publisher-  marked this subscription overflowed under-  'Kiroku.Store.Subscription.Types.DropSubscription'. Investigate the-  slow handler and either fix the slowness, raise 'queueCapacity', or-  switch to 'Kiroku.Store.Subscription.Types.DropOldest' if the consumer-  can tolerate event loss.+  marked this non-group 'Kiroku.Store.Subscription.Types.AllStreams'+  subscription overflowed under+  'Kiroku.Store.Subscription.Types.DropSubscription'. Category and+  consumer-group subscriptions do not own publisher queues, so+  'queueCapacity' and 'overflowPolicy' have no effect for them.+  Investigate the slow handler and either fix the slowness, raise+  'queueCapacity', or switch to 'Kiroku.Store.Subscription.Types.DropOldest'+  if the consumer can tolerate event loss.+* @Left e@ where @e@ is a 'Hasql.Pool.UsageError' from checkpoint load —+  startup could not read the saved checkpoint. The worker stops rather than+  silently replaying from global position 0. * @Left e@ for any exception thrown by the handler — handler exceptions   are not caught; the worker thread dies and the original exception   propagates to the consumer. This is intentional: a handler that@@ -110,65 +117,83 @@     for_ (consumerGroup config) $ \(ConsumerGroup m n) ->         when (n < 1 || m < 0 || m >= n) $             throwIO (InvalidConsumerGroup m n)-    (queue, statusVar, unsubscribe) <--        atomically $-            Pub.subscribePublisher-                (store ^. #publisher)-                (queueCapacity config)-                (overflowPolicy config)-    -- The worker writes its current FSM state here on every transition; the-    -- handle's 'currentState' reads it. Seeded with the catch-up entry state so-    -- a read before the worker's first transition is sensible.-    stateVar <- newTVarIO (CatchingUp (GlobalPosition 0) 0)-    let pubPosVar = Pub.lastPublished (store ^. #publisher)-        catGenVar = Notifier.categoryGenerations (store ^. #notifier)-    -- Register this worker's state cell into the store's central registry under-    -- (name, member). A fresh token identifies *this* worker so that an older-    -- worker's cleanup cannot delete a newer worker's replacement entry under-    -- the same key, and a held handle reads only the cell it registered.-    token <- newUnique-    let reg = store ^. #subscriptionRegistry-        key = (name config, configMember config)-        -- `finally unsubscribe` removes this subscription from the publisher's-        -- registry whenever the worker exits — gracefully on Stop, by-        -- cancellation, or on any exception (including SubscriptionOverflowed).-        -- Forgetting to unsubscribe would leave a registry entry with no-        -- reader, and the publisher would needlessly trigger this subscriber's-        -- overflow policy on the next batch. We extend the same `finally` to-        -- also deregister from the subscription-state registry on ANY exit, so-        -- the registry never leaks a stale entry. The delete is-        -- token-conditional so stale cleanup from an older duplicate-key worker-        -- cannot remove a newer worker's live entry.-        cleanup =-            unsubscribe-                >> atomically-                    ( modifyTVar' reg $-                        Map.update-                            ( \(tok', cell) ->-                                if tok' == token-                                    then Nothing-                                    else Just (tok', cell)-                            )-                            key-                    )-    -- Insert before forking so a caller that reads a snapshot immediately after-    -- `subscribe` returns already sees this subscription.-    atomically $ modifyTVar' reg (Map.insert key (token, stateVar))-    thread <--        Async.async-            ( runWorker (store ^. #pool) queue statusVar stateVar pubPosVar catGenVar config (store ^. #eventHandler) (store ^. #storeSettings)-                `finally` cleanup+    mask $ \_restore ->+        bracketOnError+            ( case (consumerGroup config, target config) of+                (Nothing, AllStreams) -> do+                    (queue, statusVar, unsubscribe) <-+                        atomically $+                            Pub.subscribePublisher+                                (store ^. #publisher)+                                (queueCapacity config)+                                (overflowPolicy config)+                    pure (LiveFromPublisherQueue queue statusVar, unsubscribe)+                (Nothing, Category (CategoryName cat)) ->+                    pure (LiveFromCategoryNotify cat, pure ())+                (Just _, _) ->+                    pure (LiveFromGroupPolling, pure ())             )-    pure-        SubscriptionHandle-            { cancel = Async.cancel thread-            , wait = Async.waitCatch thread-            , currentState = do-                m <- readTVarIO reg-                case Map.lookup key m of-                    Just (tok, cell) | tok == token -> Just <$> readTVarIO cell-                    _ -> pure Nothing-            }+            (\(_, unsubscribe) -> unsubscribe)+            $ \(liveSource, unsubscribe) -> do+                -- The worker writes its current FSM state here on every transition; the+                -- handle's 'currentState' reads it. Seeded with the catch-up entry state so+                -- a read before the worker's first transition is sensible.+                stateVar <- newTVarIO (CatchingUp (GlobalPosition 0) 0)+                let pubPosVar = Pub.lastPublished (store ^. #publisher)+                    catGenVar = Notifier.categoryGenerations (store ^. #notifier)+                -- Register this worker's state cell into the store's central registry under+                -- (name, member). A fresh token identifies *this* worker so that an older+                -- worker's cleanup cannot delete a newer worker's replacement entry under+                -- the same key, and a held handle reads only the cell it registered.+                token <- newUnique+                let reg = store ^. #subscriptionRegistry+                    key = (name config, configMember config)+                    deregister =+                        atomically+                            ( modifyTVar' reg $+                                Map.update+                                    ( \(tok', cell) ->+                                        if tok' == token+                                            then Nothing+                                            else Just (tok', cell)+                                    )+                                    key+                            )+                    -- `finally cleanup` removes this subscription from the publisher's+                    -- registry whenever the worker exits — gracefully on Stop, by+                    -- cancellation, or on any exception (including SubscriptionOverflowed).+                    -- Forgetting to unsubscribe would leave a registry entry with no+                    -- reader, and the publisher would needlessly trigger this subscriber's+                    -- overflow policy on the next batch. We extend the same `finally` to+                    -- also deregister from the subscription-state registry on ANY exit, so+                    -- the registry never leaks a stale entry. The delete is+                    -- token-conditional so stale cleanup from an older duplicate-key worker+                    -- cannot remove a newer worker's live entry.+                    cleanup = unsubscribe >> deregister+                -- Insert before forking so a caller that reads a snapshot immediately after+                -- `subscribe` returns already sees this subscription.+                bracketOnError+                    (atomically $ modifyTVar' reg (Map.insert key (token, stateVar)))+                    (const deregister)+                    $ \() -> do+                        -- Every acquisition above is paired with a release in the+                        -- pre-fork window; once asyncWithUnmask returns, ownership of+                        -- both releases has transferred to the worker thread's cleanup.+                        thread <-+                            Async.asyncWithUnmask $ \unmask ->+                                unmask+                                    (runWorker (store ^. #pool) liveSource stateVar pubPosVar catGenVar config (store ^. #eventHandler) (store ^. #storeSettings))+                                    `finally` cleanup+                        pure+                            SubscriptionHandle+                                { cancel = Async.cancel thread+                                , wait = Async.waitCatch thread+                                , currentState = do+                                    m <- readTVarIO reg+                                    case Map.lookup key m of+                                        Just (tok, cell) | tok == token -> Just <$> readTVarIO cell+                                        _ -> pure Nothing+                                }  {- | Bracket-style subscription lifecycle. 

@@ -1,12 +1,14 @@-{- | The centralized @$all@ broadcaster shared by every all-stream subscription.+{- | The centralized @$all@ broadcaster shared by queue-consuming live clients. -A single 'EventPublisher' reads new events from the global stream once and-fans them out to each registered 'Subscriber' through that subscriber's own-bounded 'Control.Concurrent.STM.TBQueue', rather than having every subscription-poll the database independently. 'startPublisher' launches the broadcast loop,+A single 'EventPublisher' reads new events from the global stream once when at+least one 'Subscriber' is registered and fans them out through each+subscriber's bounded 'Control.Concurrent.STM.TBQueue'. When the registry is+empty, it does not fetch full event rows; it advances 'lastPublished' from the+@$all@ tail with a single-row query so DB-driven subscriptions can still use the+position as their live gate. 'startPublisher' launches the loop, 'subscribePublisher' registers a subscriber (returning its queue, a status 'TVar', and an unsubscribe action), 'publisherPosition' reports the last-broadcast cursor, and 'stopPublisher' tears the loop down.+published cursor, and 'stopPublisher' tears the loop down.  When a subscriber's queue fills, the publisher applies that subscriber's 'Kiroku.Store.Subscription.Types.OverflowPolicy' by setting its@@ -49,7 +51,8 @@     writeTVar,  ) import Control.Concurrent.STM qualified as STM-import Control.Exception (throwIO)+import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)+import Control.Monad (when) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Foldable (for_) import Data.Int (Int32, Int64)@@ -60,18 +63,17 @@ import Hasql.Pool (Pool) import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session-import Kiroku.Store.Observability (KirokuEvent (..))+import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (StoreSettings, decodeEvents) import Kiroku.Store.Subscription.Types (OverflowPolicy (..)) import Kiroku.Store.Types (GlobalPosition (..), RecordedEvent (..)) import Numeric.Natural (Natural) -{- | The centralized EventPublisher. Reads events from the database once per-notification and fans them out to a registry of bounded per-subscriber-queues. The registry replaces the prior unbounded broadcast 'TChan' so-that one slow subscriber cannot grow the publisher's fan-out memory-without limit (EP-3 F6).+{- | The centralized EventPublisher. With registered subscribers, reads events+from the database once per notification and fans them out to a registry of+bounded per-subscriber queues. With an empty registry, fetches only the current+@$all@ tail to keep 'lastPublished' moving without decoding event rows. -} data EventPublisher = EventPublisher     { subscribers :: !(TVar (IntMap Subscriber))@@ -121,10 +123,11 @@ safetyPollMicros = 30_000_000  {- | Start the EventPublisher. Spawns a thread that waits for ticks from the-Notifier (or a 30-second safety poll timeout), queries the database for-new events, and delivers them to every active subscriber. Per-subscriber-overflow handling is determined by each subscription's-'OverflowPolicy'.+Notifier (or a 30-second safety poll timeout). With at least one registered+subscriber, the thread queries full event rows and delivers them to every active+subscriber; with an empty registry, it advances 'lastPublished' from the @$all@+tail using a single-row query and fetches no payload rows. Per-subscriber+overflow handling is determined by each subscription's 'OverflowPolicy'.  If an 'eventHandler' callback is supplied, the publisher emits 'Kiroku.Store.Observability.KirokuEventPublisherPoolError' when its read@@ -147,11 +150,12 @@ startPublisher pool notifierChan mHandler stSettings = liftIO $ do     subsVar <- newTVarIO IntMap.empty     nextIdVar <- newTVarIO 0+    -- Duplicate the notifier channel before reading the tail so ticks arriving+    -- during startup are redundant rather than lost until the safety poll.+    tickChan <- atomically (dupTChan notifierChan)     tailResult <- Pool.use pool (Session.statement () SQL.currentGlobalPositionStmt)     tailPos <- either throwIO pure tailResult     pos <- newTVarIO (GlobalPosition tailPos)-    -- Get a personal copy of the notifier's broadcast channel-    tickChan <- atomically (dupTChan notifierChan)     thread <- Async.async (publisherLoop pool tickChan subsVar pos mHandler stSettings)     pure         EventPublisher@@ -212,18 +216,51 @@         waitForWakeup tickChan         -- Drain all pending ticks (debouncing)         drainTicks tickChan-        -- Fetch and broadcast-        fetchAndBroadcast+        -- Fetch and broadcast. If a synchronous callback failure occurs after+        -- partial delivery, lastPublished is not advanced, so the next tick+        -- re-fetches under Kiroku's at-least-once delivery contract.+        fetchAndBroadcast `catch` \(e :: SomeException) ->+            case asyncExceptionFromException e of+                Just (ae :: SomeAsyncException) -> throwIO ae+                Nothing -> emitOrDrop mHandler (KirokuEventPublisherLoopError e)         loop      fetchAndBroadcast = do+        subs <- readTVarIO subsVar+        if IntMap.null subs+            then cheapAdvance+            else fullFetch++    cheapAdvance = do+        result <- Pool.use pool (Session.statement () SQL.currentGlobalPositionStmt)+        case result of+            Left err -> do+                -- Surface the pool error so operators see why the publisher+                -- position has stalled; the 30-second safety poll will retry.+                emitOrDrop mHandler (KirokuEventPublisherPoolError err)+            Right tailPos -> do+                -- Re-check registry emptiness in the same STM transaction as+                -- the position write. If a queue subscriber registered after+                -- our snapshot, fall through to a full fetch so no event skips+                -- a queue that now exists.+                raced <- atomically $ do+                    subs' <- readTVar subsVar+                    if IntMap.null subs'+                        then do+                            GlobalPosition cur <- readTVar posVar+                            writeTVar posVar (GlobalPosition (max cur tailPos))+                            pure False+                        else pure True+                when raced fullFetch++    fullFetch = do         GlobalPosition pos <- readTVarIO posVar         result <- Pool.use pool (Session.statement (pos, publisherBatchSize) SQL.readAllForwardStmt)         case result of             Left err -> do                 -- Surface the pool error so operators see why broadcast                 -- has stalled; the 30-second safety poll will retry.-                for_ mHandler ($ KirokuEventPublisherPoolError err)+                emitOrDrop mHandler (KirokuEventPublisherPoolError err)             Right rawEvents                 | V.null rawEvents -> pure ()                 | otherwise -> do@@ -239,18 +276,26 @@                     -- cannot rollback another's enqueue.                     subs <- readTVarIO subsVar                     for_ (IntMap.elems subs) (deliverBatch events)-                    -- Advance posVar after attempting delivery to all subscribers.-                    -- Subscribers under DropSubscription have already been marked-                    -- Overflowed; their workers will observe and exit. Subscribers-                    -- under DropOldest have lost the oldest batch but their queues-                    -- are not full anymore.-                    atomically (writeTVar posVar newPos)+                    -- Advance posVar only after attempting delivery to the+                    -- snapshot cohort. In the same transaction, re-read the+                    -- registry and offer the in-flight batch to subscribers that+                    -- registered after the snapshot. This closes the attach race:+                    -- every subscriber either received the batch before the+                    -- advance or registered after the advanced position was+                    -- visible and will cover the range through SQL catch-up.+                    atomically $ do+                        subs' <- readTVar subsVar+                        for_ (IntMap.elems (subs' `IntMap.difference` subs)) (deliverBatchSTM events)+                        writeTVar posVar newPos                     -- If we got a full batch, there may be more — loop immediately                     if V.length events >= fromIntegral publisherBatchSize-                        then fetchAndBroadcast+                        then fullFetch                         else pure ()      deliverBatch events sub = atomically $ do+        deliverBatchSTM events sub++    deliverBatchSTM events sub = do         full <- isFullTBQueue (subQueue sub)         if not full             then do

@@ -122,7 +122,7 @@       DeadLetterPoison !Text     | -- | The event payload failed validation/parsing; the 'Text' is the detail.       DeadLetterInvalid !Text-    | -- | The bounded retry budget was exhausted after the given number of attempts.+    | -- | The bounded retry budget was exhausted after the given number of total deliveries.       DeadLetterMaxAttempts !Int     | -- | A custom reason: a summary plus structured JSON detail.       DeadLetterOther !Text !Value

@@ -20,6 +20,10 @@ ('Kiroku.Store.Subscription.Types.eventTypeFilter' and 'Kiroku.Store.Subscription.Types.selector'): a filtered-out event never reaches either bridge.++Both streams end normally when the subscription worker stops cleanly or is+cancelled. If the worker dies for any other reason, the next stream pull+rethrows the worker's exception to the stream consumer. -} module Kiroku.Store.Subscription.Stream (     -- * Plain pull stream@@ -27,21 +31,32 @@      -- * Ack-coupled pull stream     AckItem (..),+    InvalidStreamBufferSize (..),     subscriptionAckStream, ) where +import Control.Concurrent.Async qualified as Async import Control.Concurrent.STM (+    STM,     TBQueue,     TMVar,+    TVar,     atomically,     newEmptyTMVarIO,     newTBQueueIO,+    newTVarIO,+    orElse,     putTMVar,     readTBQueue,+    readTVar,+    retry,     takeTMVar,     writeTBQueue,+    writeTVar,  )+import Control.Exception (Exception, SomeException, fromException, throwIO)+import Control.Monad (when) import Data.IORef (atomicModifyIORef', newIORef) import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Subscription (subscribe)@@ -77,11 +92,33 @@     -- ^ one-shot reply the consumer must fill exactly once     } +{- | Thrown by 'subscriptionAckStream' when the requested bridge queue capacity+is zero.++A zero-capacity 'TBQueue' would make the bridge handler block forever on its+first delivery, before a stream consumer can ever see the event or reply to it.+-}+newtype InvalidStreamBufferSize = InvalidStreamBufferSize Natural+    deriving stock (Eq, Show)+    deriving anyclass (Exception)++data BridgeTermination+    = BridgeClosedCleanly+    | BridgeCrashed !SomeException++closeBridge :: TVar (Maybe BridgeTermination) -> BridgeTermination -> STM ()+closeBridge closedVar termination =+    readTVar closedVar >>= \case+        Nothing -> writeTVar closedVar (Just termination)+        Just _ -> pure ()+ {- | Create a pull-based 'Stream' from a kiroku subscription.  A bounded 'TBQueue' sits between kiroku's push-based handler and the-returned Streamly stream. The stream terminates when the underlying-subscription ends or the returned cancel action runs.+returned Streamly stream. The stream terminates normally when the underlying+subscription stops cleanly or the returned cancel action runs. If the+underlying worker crashes, the next stream pull rethrows the worker exception+to the consumer.  The @handler@ field in the provided 'SubscriptionConfig' is ignored — the bridge provides its own handler.@@ -98,13 +135,13 @@ element type is unchanged ('RecordedEvent') — filtering only removes elements, it does not reshape them. -The returned cancel action cancels the underlying subscription and-writes the sentinel so any blocked reader wakes up and terminates.+The returned cancel action cancels the underlying subscription and wakes any+blocked reader without writing to the bounded queue. -} subscriptionStream ::     KirokuStore ->     SubscriptionConfig ->-    -- | TBQueue capacity (buffer size for backpressure)+    -- | TBQueue capacity for the bridge; must be at least 1.     Natural ->     IO (Stream IO RecordedEvent, IO ()) subscriptionStream store config bufferSize = do@@ -128,17 +165,23 @@ installs its own.  The returned cancel action cancels the underlying subscription (interrupting any-worker blocked waiting for a reply) and writes the @Nothing@ sentinel so a-blocked reader wakes up and the stream terminates.+worker blocked waiting for a reply) and wakes any blocked reader without+writing to the bounded queue. A clean worker stop or cancellation ends the+stream normally; any other worker exception is rethrown from the next stream+pull. This includes overflow, handler exceptions, dead-letter database errors,+and decode-hook exceptions. -} subscriptionAckStream ::     KirokuStore ->     SubscriptionConfig ->-    -- | TBQueue capacity (buffer size for backpressure)+    -- | TBQueue capacity for the bridge; must be at least 1.     Natural ->     IO (Stream IO AckItem, IO ()) subscriptionAckStream store config bufferSize = do+    when (bufferSize < 1) $+        throwIO (InvalidStreamBufferSize bufferSize)     queue <- newTBQueueIO bufferSize+    closedVar <- newTVarIO Nothing     -- Tracks the previous (eventId, attempt) so a consecutive redelivery of the     -- same event (the worker's bounded retry) is reported with an incremented     -- attempt. The worker delivers events one at a time and blocks on the reply,@@ -153,22 +196,33 @@                         | eid == eventId event -> (Just (eid, n + 1), n + 1)                     _ -> (Just (eventId event, 0), 0)             reply <- newEmptyTMVarIO-            atomically (writeTBQueue queue (Just (AckItem event attempt reply)))+            atomically (writeTBQueue queue (AckItem event attempt reply))             atomically (takeTMVar reply)      let bridgeConfig = config{handler = bridgeHandler}      subHandle <- subscribe store bridgeConfig+    _monitor <- Async.async $ do+        outcome <- wait subHandle+        atomically . closeBridge closedVar $ case outcome of+            Right () -> BridgeClosedCleanly+            Left e+                | Just Async.AsyncCancelled <- fromException e -> BridgeClosedCleanly+                | otherwise -> BridgeCrashed e      let cancelAction = do             cancel subHandle-            atomically (writeTBQueue queue Nothing)+            atomically (closeBridge closedVar BridgeClosedCleanly)      let step :: () -> IO (Maybe (AckItem, ()))         step () = do-            mItem <- atomically (readTBQueue queue)-            case mItem of-                Just item -> pure (Just (item, ()))-                Nothing -> pure Nothing+            next <-+                atomically $+                    (Right <$> readTBQueue queue)+                        `orElse` (readTVar closedVar >>= maybe retry (pure . Left))+            case next of+                Right item -> pure (Just (item, ()))+                Left BridgeClosedCleanly -> pure Nothing+                Left (BridgeCrashed e) -> throwIO e      pure (Stream.unfoldrM step (), cancelAction)

@@ -152,9 +152,10 @@       Stop     | {- | Redeliver /this same event/ after the given delay before the checkpoint       advances past it. Redelivery is bounded by the subscription's-      'RetryPolicy' ('retryMaxAttempts'); once the bound is reached the worker-      dead-letters the event with 'DeadLetterMaxAttempts' and advances to the-      next event. The adapter maps Shibuya's @AckRetry@ to this.+      'RetryPolicy' ('retryMaxAttempts'), which counts total deliveries, not+      redeliveries; once the bound is reached the worker dead-letters the event+      with 'DeadLetterMaxAttempts' and advances to the next event. The adapter+      maps Shibuya's @AckRetry@ to this.       -}       Retry !RetryDelay     | {- | Record this event in @kiroku.dead_letters@ with the given reason and@@ -164,22 +165,24 @@       DeadLetter !DeadLetterReason     deriving stock (Eq, Show) -{- | Bounds on how many times a single event is redelivered before it is+{- | Bounds how many total deliveries a single event gets before it is dead-lettered. The per-redelivery delay is carried by each 'Retry' result (mirroring Shibuya's per-decision @AckRetry RetryDelay@), so the policy only carries the attempt bound. -} newtype RetryPolicy = RetryPolicy     { retryMaxAttempts :: Int-    {- ^ Maximum redeliveries of one event before dead-lettering it. A value-    @<= 1@ means the first 'Retry' immediately dead-letters (no redelivery).+    {- ^ Maximum total deliveries of one event before dead-lettering it. The+    first delivery counts as attempt 1. A value @<= 1@ means the first 'Retry'+    immediately dead-letters (no redelivery).     -}     }     deriving stock (Eq, Show) -{- | The default 'RetryPolicy': up to five redeliveries of a single event before-it is dead-lettered with 'DeadLetterMaxAttempts'. Handlers that never return-'Retry' are unaffected by this default.+{- | The default 'RetryPolicy': up to five total deliveries of a single event —+the first delivery plus four redeliveries — before it is dead-lettered with+'DeadLetterMaxAttempts'. Handlers that never return 'Retry' are unaffected by+this default. -} defaultRetryPolicy :: RetryPolicy defaultRetryPolicy = RetryPolicy{retryMaxAttempts = 5}@@ -244,17 +247,21 @@     -- ^ Number of events to fetch per batch during catch-up (default: 100)     , queueCapacity :: !Natural     {- ^ Maximum number of /batches/ the publisher may enqueue for this-    subscriber before applying 'overflowPolicy'. Each batch is up to+    subscriber before applying 'overflowPolicy'. Applies only to non-group+    'AllStreams' subscriptions, because Category and consumer-group+    subscriptions do not create publisher queues. Each batch is up to     'EventPublisher.publisherBatchSize' events, so the effective event-    capacity is @queueCapacity * publisherBatchSize@. Default: 16-    batches (~16,000 events at the default publisher batch size).+    capacity is @queueCapacity * publisherBatchSize@. Default: 16 batches+    (~16,000 events at the default publisher batch size).     -}     , overflowPolicy :: !OverflowPolicy     {- ^ What the publisher does when this subscriber's queue is full.-    Default: 'PauseAndResume' — a slow subscriber is paused and then-    recovers losslessly (re-reading missed events from its checkpoint)-    rather than being terminated or silently growing the publisher's-    fan-out memory.+    Applies only to non-group 'AllStreams' subscriptions, because Category+    and consumer-group subscriptions are DB-driven and have no publisher+    queue. Default: 'PauseAndResume' — a slow subscriber is paused and+    then recovers losslessly (re-reading missed events from its checkpoint)+    rather than being terminated or silently growing the publisher's fan-out+    memory.     -}     , consumerGroup :: !(Maybe ConsumerGroup)     {- ^ 'Nothing' (the default) = ordinary single-consumer subscription.

@@ -18,12 +18,15 @@ the single delivery primitive shared by every live path, so behaviour is identical for @AllStreams@, @Category@, and consumer-group subscriptions. -'withFetchBatchHookForTest' is a test-only seam for injecting fetch failures.+'withFetchBatchHookForTest' and 'withLoadCheckpointHookForTest' are test-only+seams for injecting fetch and checkpoint-load failures. -} module Kiroku.Store.Subscription.Worker (+    LiveSource (..),     runWorker,     configMember,     withFetchBatchHookForTest,+    withLoadCheckpointHookForTest, ) where  import Contravariant.Extras (contrazip2)@@ -32,7 +35,6 @@ import Control.Concurrent.STM (TBQueue, TVar, atomically, check, orElse, readTBQueue, readTVar, registerDelay, tryReadTBQueue, writeTVar) import Control.Exception (SomeException, bracket, fromException, throwIO, try) import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.Foldable (for_) import Data.IORef (IORef, newIORef, readIORef, writeIORef) import Data.Int (Int32, Int64) import Data.Map.Strict (Map)@@ -53,6 +55,7 @@     SubscriptionDeliveryPhase (..),     SubscriptionGroupContext (..),     SubscriptionStopReason (..),+    emitOrDrop,  ) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (StoreSettings, decodeEvents)@@ -81,10 +84,18 @@     GlobalPosition ->     IO (Maybe (Either Pool.UsageError (Vector RecordedEvent))) +type LoadCheckpointHook =+    SubscriptionConfig ->+    IO (Maybe (Either Pool.UsageError (Maybe Int64)))+ {-# NOINLINE fetchBatchHookRef #-} fetchBatchHookRef :: IORef (Maybe FetchBatchHook) fetchBatchHookRef = unsafePerformIO (newIORef Nothing) +{-# NOINLINE loadCheckpointHookRef #-}+loadCheckpointHookRef :: IORef (Maybe LoadCheckpointHook)+loadCheckpointHookRef = unsafePerformIO (newIORef Nothing)+ {- | Install a process-local fetch hook for tests that need deterministic subscription-worker fault injection. Production code leaves the hook unset. -}@@ -99,16 +110,51 @@         (writeIORef fetchBatchHookRef)         (const action) +{- | Install a process-local checkpoint-load hook for tests that need+deterministic subscription-startup fault injection. Production code leaves the+hook unset.+-}+withLoadCheckpointHookForTest :: LoadCheckpointHook -> IO a -> IO a+withLoadCheckpointHookForTest hook action =+    bracket+        ( do+            previous <- readIORef loadCheckpointHookRef+            writeIORef loadCheckpointHookRef (Just hook)+            pure previous+        )+        (writeIORef loadCheckpointHookRef)+        (const action)+ fetchRetryDelayMicros :: Int -> Int fetchRetryDelayMicros attempt =     min categorySafetyPollMicros (100_000 * (2 ^ min attempt 9 :: Int)) +{- | How a worker obtains live-mode batches, fixed at 'subscribe' time from the+config's (consumerGroup, target) shape.++Only 'LiveFromPublisherQueue' owns a registration with the EventPublisher. The+other shapes are DB-driven and the publisher must do no fan-out work for them.+-}+data LiveSource+    = {- | Non-group AllStreams: read the publisher's bounded queue; the status+      TVar carries Paused/Overflowed backpressure signals.+      -}+      LiveFromPublisherQueue !(TBQueue (Vector RecordedEvent)) !(TVar SubscriberStatus)+    | {- | Non-group Category: wake on the named category's NOTIFY generation+      counter and re-query the database.+      -}+      LiveFromCategoryNotify !Text+    | {- | Consumer-group member, for either target: wake when the global+      position advances and re-query with the partition predicate.+      -}+      LiveFromGroupPolling+ {- | Run the subscription worker loop. Two phases:  Phase 1 (catch-up): queries database directly until reaching publisherPosition.-Phase 2 (live): for AllStreams, reads from the bounded TBQueue the publisher-delivers to; for Category, re-queries the database whenever the publisher-advances (the broadcast TBQueue is unused for category subscriptions).+Phase 2 (live): for 'LiveFromPublisherQueue', reads from the bounded TBQueue+the publisher delivers to; for 'LiveFromCategoryNotify' and+'LiveFromGroupPolling', re-queries the database, and no publisher queue exists.  Runs until the handler returns 'Stop', the thread is cancelled, or the publisher signals overflow on the subscriber's status TVar (in which@@ -132,8 +178,7 @@ runWorker ::     (MonadIO m) =>     Pool ->-    TBQueue (Vector RecordedEvent) ->-    TVar SubscriberStatus ->+    LiveSource ->     {- | the worker's current FSM state, written on every transition so callers     can read it through 'Kiroku.Store.Subscription.Types.currentState'.     -}@@ -152,8 +197,8 @@     -}     StoreSettings ->     m ()-runWorker pool liveQueue statusVar stateVar pubPosVar catGenVar config mHandler stSettings = liftIO $ do-    let emit evt = for_ mHandler ($ evt)+runWorker pool liveSource stateVar pubPosVar catGenVar config mHandler stSettings = liftIO $ do+    let emit = emitOrDrop mHandler         subName = name config         groupCtx = groupCtxOf config     posRef <- newIORef (GlobalPosition 0)@@ -225,8 +270,8 @@                             Right events                                 | V.null events -> pure CaughtUp                                 | otherwise -> pure (BatchFetched events)-            Live c -> case (consumerGroup config, target config) of-                (Nothing, AllStreams) -> do+            Live c -> case liveSource of+                LiveFromPublisherQueue liveQueue statusVar -> do                     writeIORef posRef c                     atomically $ do                         status <- readTVar statusVar@@ -242,9 +287,9 @@                                 events <- readTBQueue liveQueue                                 let fresh = V.filter ((> c) . globalPosition) events                                 pure (if V.null fresh then FetchEmpty else BatchFetched fresh)-                (Nothing, Category (CategoryName cat)) ->+                LiveFromCategoryNotify cat ->                     liveExitToInput =<< liveLoopCategoryNotify pool config stateVar catGenVar cat emit posRef c stSettings-                (Just _, _) ->+                LiveFromGroupPolling ->                     liveExitToInput =<< liveLoopDbDriven pool config stateVar pubPosVar emit posRef c stSettings             -- Recoverable backpressure: the publisher set 'Paused' because this             -- subscriber's bounded queue filled. Drain the stale queue (those@@ -254,10 +299,17 @@             -- AllStreams live path's @> cursor@ filter drops any superseded queued             -- entries once the worker is live again.             Paused{} -> do-                atomically $ do-                    drainQueue liveQueue-                    writeTVar statusVar Pub.Active-                pure QueueDrained+                case liveSource of+                    LiveFromPublisherQueue liveQueue statusVar -> do+                        atomically $ do+                            drainQueue liveQueue+                            writeTVar statusVar Pub.Active+                        pure QueueDrained+                    -- Defensive totality: only the queue branch can produce+                    -- QueueBackpressured, so DB-driven sources should never+                    -- enter Paused.+                    LiveFromCategoryNotify{} -> pure QueueDrained+                    LiveFromGroupPolling -> pure QueueDrained             -- Reconnecting: re-probe the database from the checkpoint. A success             -- re-enters catch-up (delivering everything after the cursor); a             -- failure stays in 'Reconnecting' for another backed-off attempt; an@@ -383,10 +435,9 @@ configMember :: SubscriptionConfig -> Int32 configMember config = maybe 0 member (consumerGroup config) --- Load the checkpoint from the database, defaulting to 0.--- Surfaces a database error through the event handler before falling--- back to the safe default; without the event the operator has no signal--- that a transient pool error caused a silent re-process from position 0.+-- Load the checkpoint from the database, defaulting to 0 only when no checkpoint+-- row exists. A database error is emitted and rethrown so startup fails loudly+-- instead of silently re-processing from position 0. -- Keyed by (subscription_name, member) so each group member resumes from its -- own saved position. loadCheckpoint ::@@ -397,11 +448,15 @@ loadCheckpoint pool config emit = do     let subName@(SubscriptionName name') = name config         mem = configMember config-    result <- Pool.use pool (Session.statement (name', mem) SQL.getCheckpointMemberStmt)+    mHook <- readIORef loadCheckpointHookRef+    injected <- maybe (pure Nothing) (\hook -> hook config) mHook+    result <- case injected of+        Just hooked -> pure hooked+        Nothing -> Pool.use pool (Session.statement (name', mem) SQL.getCheckpointMemberStmt)     case result of         Left err -> do             emit (KirokuEventSubscriptionDbError subName LoadCheckpoint err (groupCtxOf config))-            pure (GlobalPosition 0)+            throwIO err         Right Nothing -> pure (GlobalPosition 0)         Right (Just pos) -> pure (GlobalPosition pos) 

@@ -70,7 +70,7 @@ import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Effect (PreparedEvent, Store (..), appendDispatchTx, buildAppendParams, prepareEvents) import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore)-import Kiroku.Store.Error (AppendConflict (..), StoreError (..), appendConflictToStoreError, emptyResultConflict)+import Kiroku.Store.Error (AppendConflict (..), StoreError (..), appendConflictToStoreError, emptyResultConflict, validateStreamName) import Kiroku.Store.Settings qualified as Settings import Kiroku.Store.Types @@ -137,6 +137,15 @@ entering the transaction body, or use 'runTransactionAppending', which performs the rejection up front. +Lock-hold note: this append statement updates Kiroku's global @$all@+stream row and PostgreSQL holds that row lock until the surrounding+transaction commits or rolls back. Any statements the caller runs+after a successful 'appendToStreamTx' keep that global append lock+held, blocking other appends store-wide. Prefer the higher-level+'runTransactionAppending' wrappers unless you need custom transaction+shaping; when using this function directly, run any append-independent+work before calling it.+ The combinator returns 'Either' instead of throwing because 'Tx.Transaction' has no exception channel. On 'Left' the caller decides whether to call 'Tx.condemn' to roll back the transaction, branch@@ -147,8 +156,10 @@ generates UUIDv7s) and the @createdAt@ timestamp must be captured by the caller prior to entering the transaction body. -Empty event lists are a programming mistake — see-'Kiroku.Store.Append.appendToStream' for the rationale.+Empty event lists are a programming mistake. This function returns+@'Left' ('EmptyAppendBatchConflict' …)@ without dispatching a statement;+the high-level @runTransactionAppending*@ wrappers return+@'Left' ('EmptyAppendBatch' …)@ before opening a transaction. -} appendToStreamTx ::     StreamName ->@@ -157,11 +168,15 @@     UTCTime ->     Tx.Transaction (Either AppendConflict AppendResult) appendToStreamTx sn@(StreamName name) expected prepared now = do-    let params = buildAppendParams name now prepared-    mResult <- appendDispatchTx expected params-    pure $ case mResult of-        Just r -> Right r-        Nothing -> Left (emptyResultConflict sn expected)+    case prepared of+        [] ->+            pure (Left (EmptyAppendBatchConflict sn))+        _ -> do+            let params = buildAppendParams name now prepared+            mResult <- appendDispatchTx expected params+            pure $ case mResult of+                Just r -> Right r+                Nothing -> Left (emptyResultConflict sn expected)  -- --------------------------------------------------------------------------- -- Convenience wrapper@@ -176,9 +191,13 @@  Behavior: -* If the target stream is the reserved @$all@, the call rejects with-  @'Left' ('ReservedStreamName' …)@ /before/ opening any transaction;-  the continuation is not invoked.+* If the target stream name is invalid (reserved @$all@ or longer than+  'Kiroku.Store.Error.maxStreamNameBytes'), the call rejects with+  @'Left'@ /before/ opening any transaction; the continuation is not+  invoked.+* If the event list is empty, the call rejects with+  @'Left' ('EmptyAppendBatch' …)@ /before/ opening any transaction; the+  continuation is not invoked. * UUIDv7 ids are generated for events with a 'Nothing' 'eventId', and   the current time is captured, before the transaction begins   ('Tx.Transaction' has no 'MonadIO' so this prep cannot happen@@ -195,6 +214,22 @@   a serialization conflict; use 'runTransactionAppendingNoRetry' to   disable retry. +Lock-hold guidance:++* The append updates Kiroku's global @$all@ stream row. PostgreSQL+  holds that row lock until this transaction commits or rolls back.+* The continuation runs /after/ the append, so every SQL statement it+  executes extends the interval during which all other appends in the+  store are blocked on the @$all@ row.+* Keep the continuation minimal: pre-compute values before entering+  the transaction, avoid unbounded loops or per-row work that could+  have been batched, and never wait on external systems. A+  'Tx.Transaction' cannot perform arbitrary 'IO', but it can still+  contain many database round trips.+* Work that does not need the new 'AppendResult' should run before the+  append, either outside this wrapper or in a custom 'runTransaction'+  body that calls 'appendToStreamTx' last.+ Connection-level failures from the @hasql-pool@ layer surface through the surrounding @'Effectful.Error.Static.Error' 'StoreError'@ effect ('Kiroku.Store.Error.ConnectionError', 'ConnectionLost', etc.), not@@ -227,7 +262,8 @@  Like 'runTransactionAppending', this wrapper bypasses the 'Kiroku.Store.Settings.enrichEvent' hook; see that function's-Haddock for guidance on hooking under different stacks.+Haddock for guidance on hooking under different stacks and for the+@$all@ lock-hold warning that applies equally here. -} runTransactionAppendingNoRetry ::     (HasCallStack, IOE :> es, Store :> es) =>@@ -250,7 +286,9 @@ This is the recommended path for callers — most prominently the @keiro@ projection layer — that want trace-context or PII-injection hooks to fire uniformly across both direct @'appendToStream'@ calls-and transactional append+projection sites.+and transactional append+projection sites. The continuation still runs+after the append; see 'runTransactionAppending' for the @$all@+lock-hold warning and keep resource-aware continuations just as small. -} runTransactionAppendingResource ::     (HasCallStack, IOE :> es, KirokuStoreResource :> es, Store :> es) =>@@ -265,7 +303,9 @@     runTransactionAppendingWith RunTransaction sn expected events' k  {- | Like 'runTransactionAppendingResource' but uses-'Hasql.Transaction.Sessions.transactionNoRetry' under the hood.+'Hasql.Transaction.Sessions.transactionNoRetry' under the hood. The+@$all@ lock-hold guidance from 'runTransactionAppending' applies+unchanged. -} runTransactionAppendingResourceNoRetry ::     (HasCallStack, IOE :> es, KirokuStoreResource :> es, Store :> es) =>@@ -306,7 +346,8 @@     (AppendResult -> Tx.Transaction a) ->     Eff es (Either StoreError a) runTransactionAppendingWith ctor sn@(StreamName name) expected events k-    | name == "$all" = pure (Left (ReservedStreamName sn))+    | Left err <- validateStreamName sn = pure (Left err)+    | null events = pure (Left (EmptyAppendBatch sn))     | otherwise = do         prepared <- prepareEventsIO events         now <- liftIO getCurrentTime

@@ -12,12 +12,15 @@     AppendResult (..),     LinkResult (..),     CategoryName (..),+    categoryName,+    streamNameInCategory,     EventFilter (..), ) where  import Data.Aeson (Value) import Data.Int (Int64) import Data.Text (Text)+import Data.Text qualified as Text import Data.Time (UTCTime) import Data.UUID (UUID) import GHC.Generics (Generic)@@ -79,9 +82,16 @@ newtype StreamVersion = StreamVersion Int64     deriving stock (Eq, Ord, Show, Generic) -{- | Monotonic global position counter, shared across all streams.-Strictly increasing per successful append; gap-free under all four-'ExpectedVersion' paths (see EP-1's audit). Used by the @$all@ read+{- | Global position of an event in the @$all@ ordering, shared across all+streams. __Contract:__ strictly increasing per successful append, and totally+ordered — nothing more. Treat values as opaque cursors: construct a+'GlobalPosition' only from @0@ (the beginning of the store) or from a value+previously returned by this store; never derive one by arithmetic, and never+assume positions are dense (@pos + 1@ may not exist). The current+implementation happens to assign contiguous positions (see EP-1's audit), but+contiguity is an implementation detail, not an API guarantee, and is the part+that would change under a sequence-based allocation scheme — see+docs/architecture/global-position-migration-path.md. Used by the @$all@ read path and by subscriptions as the cursor key. -} newtype GlobalPosition = GlobalPosition Int64@@ -182,6 +192,13 @@     soft-deleted, 'Nothing' otherwise. Hard-deleted streams do not     appear in 'getStream' at all.     -}+    , truncateBefore :: !StreamVersion+    {- ^ The logical truncate-before marker. Ordered per-stream reads+    ('Kiroku.Store.Read.readStreamForward' / 'readStreamBackward') return only+    events whose per-stream version is >= this value. 0 (the default) keeps the+    whole stream. Does not affect the @$all@ global log, category reads, or+    subscriptions. Set with 'Kiroku.Store.Lifecycle.setStreamTruncateBefore'.+    -}     }     deriving stock (Eq, Show, Generic) @@ -266,6 +283,38 @@ -} newtype CategoryName = CategoryName Text     deriving stock (Eq, Ord, Show, Generic)++{- | The category of a stream name: the substring before the first @-@. This+is the canonical Haskell mirror of the+@streams.category GENERATED ALWAYS AS split_part(stream_name,'-',1)@ column and+of the category rule used by 'Kiroku.Store.Read.readCategory' and+category-targeted subscriptions. A category never contains @-@, so this is+exact.++>>> categoryName (StreamName "orders-1")+CategoryName "orders"++A name without a dash is its own category:++>>> categoryName (StreamName "singleton")+CategoryName "singleton"+-}+categoryName :: StreamName -> CategoryName+categoryName (StreamName t) = CategoryName (Text.takeWhile (/= '-') t)++{- | Build the stream name for a category and an id segment, i.e.+@\<category\>-\<id\>@. The dual of 'categoryName': for any dash-free category,+@'categoryName' ('streamNameInCategory' cat seg) == cat@.++This is permissive — it does not reject a hyphenated 'CategoryName' or an empty+segment (the store accepts such names); opinionated validation is the caller's+concern.++>>> streamNameInCategory (CategoryName "orders") "1"+StreamName "orders-1"+-}+streamNameInCategory :: CategoryName -> Text -> StreamName+streamNameInCategory (CategoryName cat) seg = StreamName (cat <> "-" <> seg)  {- | Filter passed to the 'Kiroku.Store.Effect.FindEvents' constructor and the public smart constructors in "Kiroku.Store.Causation". Each constructor maps

@@ -19,6 +19,8 @@ import Data.Vector qualified as V import Effectful (Eff, IOE, runEff, (:>)) import Effectful.Error.Static (runErrorNoCallStack)+import Hasql.Errors qualified as Errors+import Hasql.Pool (UsageError (..)) import Kiroku.Store import Kiroku.Store.Error (extractStreamNameFromDetail) import Kiroku.Store.Subscription.Effect qualified as SubEff@@ -26,6 +28,7 @@ import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..), SubscriptionOverflowed (..)) import Kiroku.Test.Postgres (withMigratedTestDatabase) import Test.CatchupDbErrorNoPrematureSwitch qualified as CatchupDbErrorNoPrematureSwitch+import Test.Category qualified as Category import Test.CategoryIdleNoSpin qualified as CategoryIdleNoSpin import Test.Causation qualified as Causation import Test.Concurrency qualified as Concurrency@@ -37,9 +40,14 @@ import Test.Helpers import Test.Hspec import Test.InterpreterHooks qualified as InterpreterHooks+import Test.NotifyGuard qualified as NotifyGuard import Test.Properties qualified as Properties+import Test.PublisherCallbackResilience qualified as PublisherCallbackResilience+import Test.PublisherIdleAdvance qualified as PublisherIdleAdvance import Test.PublisherRestartNoRebroadcast qualified as PublisherRestartNoRebroadcast import Test.ReadStream qualified as ReadStream+import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing+import Test.StreamBridgeTermination qualified as StreamBridgeTermination import Test.StreamNameLookup qualified as StreamNameLookup import Test.SubscriptionPauseResume qualified as SubscriptionPauseResume import Test.SubscriptionReconnect qualified as SubscriptionReconnect@@ -47,25 +55,33 @@ import Test.SubscriptionRetryDeadLetter qualified as SubscriptionRetryDeadLetter import Test.SubscriptionState qualified as SubscriptionState import Test.Transaction qualified as Transaction+import Test.TruncateBefore qualified as TruncateBefore  main :: IO () main = withSharedMigratedPostgres $ hspec $ do+    Category.spec     Properties.spec     Concurrency.spec     FailureInjection.spec     Transaction.spec+    TruncateBefore.spec     ReadStream.spec     StreamNameLookup.spec+    StreamBridgeTermination.spec     InterpreterHooks.spec     Causation.spec     ConsumerGroupSql.spec     ConsumerGroup.spec     ConsumerGroupEffect.spec+    NotifyGuard.spec     CategoryIdleNoSpin.spec+    PublisherCallbackResilience.spec+    PublisherIdleAdvance.spec     PublisherRestartNoRebroadcast.spec     CatchupDbErrorNoPrematureSwitch.spec     SubscriptionPauseResume.spec     SubscriptionReconnect.spec+    StartupFailureSurfacing.spec     SubscriptionState.spec     SubscriptionRegistry.spec     SubscriptionRetryDeadLetter.spec@@ -112,6 +128,43 @@                         Left (StreamAlreadyExists _) -> pure ()                         other -> expectationFailure ("Expected StreamAlreadyExists, got: " <> show other) +            describe "empty event batches" $ do+                it "rejects NoStream without creating a phantom stream or advancing $all" $ \store -> do+                    Right beforeAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 10+                    result <- runStoreIO store $ appendToStream (StreamName "empty-nostream") NoStream []+                    result `shouldBe` Left (EmptyAppendBatch (StreamName "empty-nostream"))+                    runStoreIO store (getStream (StreamName "empty-nostream")) `shouldReturn` Right Nothing+                    Right afterAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 10+                    fmap (^. #globalPosition) (V.toList afterAll) `shouldBe` fmap (^. #globalPosition) (V.toList beforeAll)++                it "rejects empty batches for existing-stream expectations without changing version" $ \store -> do+                    Right _ <- runStoreIO store $ appendToStream (StreamName "empty-existing") NoStream [makeEvent "Created" (Aeson.object [])]+                    let assertEmptyRejected expected = do+                            result <- runStoreIO store $ appendToStream (StreamName "empty-existing") expected []+                            result `shouldBe` Left (EmptyAppendBatch (StreamName "empty-existing"))+                            Right (Just info) <- runStoreIO store $ getStream (StreamName "empty-existing")+                            (info ^. #version) `shouldBe` StreamVersion 1+                    assertEmptyRejected (ExactVersion (StreamVersion 1))+                    assertEmptyRejected StreamExists+                    assertEmptyRejected AnyVersion++                it "rejects an empty per-stream batch in appendMultiStream without committing any stream" $ \store -> do+                    let ok = StreamName "empty-multi-ok"+                        bad = StreamName "empty-multi-bad"+                    result <-+                        runStoreIO store $+                            appendMultiStream+                                [ (ok, NoStream, [makeEvent "Created" (Aeson.object [])])+                                , (bad, NoStream, [])+                                ]+                    result `shouldBe` Left (EmptyAppendBatch bad)+                    runStoreIO store (getStream ok) `shouldReturn` Right Nothing+                    runStoreIO store (getStream bad) `shouldReturn` Right Nothing++                it "treats an empty appendMultiStream operation list as a no-op" $ \store -> do+                    result <- runStoreIO store $ appendMultiStream []+                    result `shouldBe` Right []+             describe "ExactVersion" $ do                 it "appends when version matches" $ \store -> do                     Right r1 <- runStoreIO store $ appendToStream (StreamName "order-789") NoStream [makeEvent "OrderCreated" (Aeson.object [])]@@ -277,6 +330,45 @@                         Just si -> (si ^. #deletedAt) `shouldBe` Nothing                         Nothing -> expectationFailure "Expected reserved $all stream row to remain present" +                it "rejects stream names longer than 512 UTF-8 bytes before creating streams" $ \store -> do+                    let overAscii = StreamName (T.replicate 513 "a")+                        overUtf8 = StreamName (T.replicate 171 (T.singleton '\12354'))+                        assertTooLong sn = do+                            result <- runStoreIO store $ appendToStream sn NoStream [makeEvent "TooLong" (Aeson.object [])]+                            result `shouldBe` Left (StreamNameTooLong sn 513)+                            runStoreIO store (getStream sn) `shouldReturn` Right Nothing+                    assertTooLong overAscii+                    assertTooLong overUtf8++                it "rejects oversized link targets without creating the target stream" $ \store -> do+                    let over = StreamName (T.replicate 513 "l")+                    Right _ <- runStoreIO store $ appendToStream (StreamName "length-link-source") NoStream [makeEvent "Source" (Aeson.object [])]+                    Right events <- runStoreIO store $ readStreamForward (StreamName "length-link-source") (StreamVersion 0) 10+                    let eid = V.head events ^. #eventId+                    result <- runStoreIO store $ linkToStream over [eid]+                    result `shouldBe` Left (StreamNameTooLong over 513)+                    runStoreIO store (getStream over) `shouldReturn` Right Nothing++                it "rejects oversized multi-stream targets without partial commit" $ \store -> do+                    let ok = StreamName "length-multi-ok"+                        over = StreamName (T.replicate 513 "m")+                    result <-+                        runStoreIO store $+                            appendMultiStream+                                [ (ok, NoStream, [makeEvent "ShouldRollback" (Aeson.object [])])+                                , (over, NoStream, [makeEvent "TooLong" (Aeson.object [])])+                                ]+                    result `shouldBe` Left (StreamNameTooLong over 513)+                    runStoreIO store (getStream ok) `shouldReturn` Right Nothing+                    runStoreIO store (getStream over) `shouldReturn` Right Nothing++                it "accepts a 512-byte stream name" $ \store -> do+                    let exact = StreamName (T.replicate 512 "e")+                    result <- runStoreIO store $ appendToStream exact NoStream [makeEvent "ExactLength" (Aeson.object [])]+                    case result of+                        Right r -> (r ^. #streamVersion) `shouldBe` StreamVersion 1+                        Left err -> expectationFailure ("Expected 512-byte stream name to append, got: " <> show err)+         describe "readStreamForward" $ do             it "reads events in forward order (read-your-own-writes)" $ \store -> do                 let events =@@ -311,6 +403,24 @@                 (result V.! 1 ^. #streamVersion) `shouldBe` StreamVersion 2                 (result V.! 2 ^. #streamVersion) `shouldBe` StreamVersion 1 +            it "paginates backward using stream version as an exclusive cursor" $ \store -> do+                let events = map (\i -> makeEvent ("E" <> T.pack (show i)) (Aeson.object [])) [1 .. 5 :: Int]+                Right _ <- runStoreIO store $ appendToStream (StreamName "read-bwd-page") NoStream events++                Right page1 <- runStoreIO store $ readStreamBackward (StreamName "read-bwd-page") (StreamVersion 0) 2+                fmap (^. #streamVersion) (V.toList page1) `shouldBe` [StreamVersion 5, StreamVersion 4]++                let cursor1 = V.last page1 ^. #streamVersion+                Right page2 <- runStoreIO store $ readStreamBackward (StreamName "read-bwd-page") cursor1 2+                fmap (^. #streamVersion) (V.toList page2) `shouldBe` [StreamVersion 3, StreamVersion 2]++                let cursor2 = V.last page2 ^. #streamVersion+                Right page3 <- runStoreIO store $ readStreamBackward (StreamName "read-bwd-page") cursor2 2+                fmap (^. #streamVersion) (V.toList page3) `shouldBe` [StreamVersion 1]++                Right page4 <- runStoreIO store $ readStreamBackward (StreamName "read-bwd-page") (V.last page3 ^. #streamVersion) 2+                V.length page4 `shouldBe` 0+         describe "cursor-based pagination" $ do             it "paginates using stream version as cursor" $ \store -> do                 let events = map (\i -> makeEvent ("E" <> T.pack (show i)) (Aeson.object [])) [1 .. 5 :: Int]@@ -357,6 +467,20 @@                 (V.head result ^. #eventType) `shouldBe` EventType "Y"                 (result V.! 1 ^. #eventType) `shouldBe` EventType "X" +            it "paginates backward using global position as an exclusive cursor" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "allb-page-s1") NoStream [makeEvent "X" (Aeson.object [])]+                Right _ <- runStoreIO store $ appendToStream (StreamName "allb-page-s2") NoStream [makeEvent "Y" (Aeson.object [])]+                Right _ <- runStoreIO store $ appendToStream (StreamName "allb-page-s3") NoStream [makeEvent "Z" (Aeson.object [])]++                Right page1 <- runStoreIO store $ readAllBackward (GlobalPosition 0) 2+                fmap (^. #globalPosition) (V.toList page1) `shouldBe` [GlobalPosition 3, GlobalPosition 2]++                Right page2 <- runStoreIO store $ readAllBackward (GlobalPosition 2) 2+                fmap (^. #globalPosition) (V.toList page2) `shouldBe` [GlobalPosition 1]++                Right page3 <- runStoreIO store $ readAllBackward (GlobalPosition 1) 2+                V.length page3 `shouldBe` 0+         describe "read empty/nonexistent stream" $ do             it "returns empty Vector for nonexistent stream" $ \store -> do                 Right result <- runStoreIO store $ readStreamForward (StreamName "no-such-stream") (StreamVersion 0) 100@@ -532,7 +656,9 @@                 Right _ <- runStoreIO store $ linkToStream (StreamName "linked-dup") [eid]                 result <- runStoreIO store $ linkToStream (StreamName "linked-dup") [eid]                 case result of-                    Left _ -> pure () -- Expected: some error (PK violation)+                    Left (EventAlreadyLinked (StreamName "linked-dup") mEid) ->+                        mEid `shouldBe` Just eid+                    Left err -> expectationFailure ("Expected EventAlreadyLinked, got: " <> show err)                     Right _ -> expectationFailure "Expected error for duplicate link"              -- F3 regression — silent version gap when source events do not exist.@@ -544,7 +670,8 @@                 let bogus = EventId (case UUID.fromString "11111111-1111-7111-8111-111111111111" of Just u -> u; Nothing -> error "bad uuid")                 result <- runStoreIO store $ linkToStream (StreamName "f3-bogus") [bogus]                 case result of-                    Left _ -> pure ()+                    Left (LinkSourceEventMissing (StreamName "f3-bogus")) -> pure ()+                    Left err -> expectationFailure ("Expected LinkSourceEventMissing, got: " <> show err)                     Right r -> expectationFailure ("Expected error for missing event, got: " <> show r)                 -- Target stream must not have been left in a half-created state.                 Right info <- runStoreIO store $ getStream (StreamName "f3-bogus")@@ -557,7 +684,8 @@                 let bogus = EventId (case UUID.fromString "22222222-2222-7222-8222-222222222222" of Just u -> u; Nothing -> error "bad uuid")                 result <- runStoreIO store $ linkToStream (StreamName "f3-mixed-tgt") [realId, bogus]                 case result of-                    Left _ -> pure ()+                    Left (LinkSourceEventMissing (StreamName "f3-mixed-tgt")) -> pure ()+                    Left err -> expectationFailure ("Expected LinkSourceEventMissing, got: " <> show err)                     Right r -> expectationFailure ("Expected error for partial-missing batch, got: " <> show r)                 Right info <- runStoreIO store $ getStream (StreamName "f3-mixed-tgt")                 info `shouldBe` Nothing@@ -850,6 +978,39 @@                 after <- countEvents store                 (before - after) `shouldBe` 3 +            it "removes dead letters for orphaned events before deleting their payloads" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "dl-hard-source") NoStream [makeEvent "DeadLettered" (Aeson.object [])]+                Right events <- runStoreIO store $ readStreamForward (StreamName "dl-hard-source") (StreamVersion 0) 10+                let event = V.head events+                    eid = event ^. #eventId+                insertDeadLetterForEvent store "dl-hard-subscription" event+                countDeadLettersForEvents store [eid] `shouldReturn` 1++                Right mId <- runStoreIO store $ hardDeleteStream (StreamName "dl-hard-source")+                mId `shouldSatisfy` (/= Nothing)+                countDeadLettersForEvents store [eid] `shouldReturn` 0+                Right remaining <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+                V.filter (\recorded -> (recorded ^. #eventId) == eid) remaining `shouldBe` V.empty++            it "keeps dead letters for linked-in events that survive a hard delete" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "dl-origin") NoStream [makeEvent "Survivor" (Aeson.object [])]+                Right originEvents <- runStoreIO store $ readStreamForward (StreamName "dl-origin") (StreamVersion 0) 10+                let event = V.head originEvents+                    eid = event ^. #eventId+                Right _ <- runStoreIO store $ linkToStream (StreamName "dl-link-target") [eid]+                insertDeadLetterForEvent store "dl-survivor-subscription" event+                countDeadLettersForEvents store [eid] `shouldReturn` 1++                Right _ <- runStoreIO store $ hardDeleteStream (StreamName "dl-link-target")+                countDeadLettersForEvents store [eid] `shouldReturn` 1+                Right afterTargetDelete <- runStoreIO store $ readStreamForward (StreamName "dl-origin") (StreamVersion 0) 10+                fmap (^. #eventId) (V.toList afterTargetDelete) `shouldBe` [eid]++                Right _ <- runStoreIO store $ hardDeleteStream (StreamName "dl-origin")+                countDeadLettersForEvents store [eid] `shouldReturn` 0+                Right afterOriginDelete <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+                V.filter (\recorded -> (recorded ^. #eventId) == eid) afterOriginDelete `shouldBe` V.empty+             -- F1 regression — events that are linked to other streams must survive             -- hard-delete of their source stream's owner if any non-target junctions remain.             it "preserves events still linked to non-target streams" $ \store -> do@@ -1725,6 +1886,24 @@         it "returns Nothing for an empty parenthesised value" $ do             extractStreamNameFromDetail "Key (stream_name)=() already exists."                 `shouldBe` Nothing++    describe "isTransientSerializationError" $ do+        let serverUsage code =+                SessionUsageError $+                    Errors.StatementSessionError+                        1+                        0+                        ""+                        []+                        True+                        ( Errors.ServerStatementError $+                            Errors.ServerError code "server error" Nothing Nothing Nothing+                        )+        it "recognizes only serialization and deadlock SQLSTATEs" $ do+            isTransientSerializationError (serverUsage "40P01") `shouldBe` True+            isTransientSerializationError (serverUsage "40001") `shouldBe` True+            isTransientSerializationError (serverUsage "23505") `shouldBe` False+            isTransientSerializationError AcquisitionTimeoutUsageError `shouldBe` False      -- =================================================================     -- Notifier reconnection tests (EP-3 F1)

@@ -0,0 +1,70 @@+{- | Pure laws for the category accessor and safe stream-name constructor+('Kiroku.Store.Types.categoryName' and 'streamNameInCategory').++These pin the "category = substring before the first @-@" rule — enforced by+the @streams.category GENERATED ALWAYS AS split_part(stream_name,'-',1)@ column+and re-derived in 'Kiroku.Store.Notification' — so the public accessor cannot+drift from it. The round-trip law is the contract keiro's typed @Category@ API+(ExecPlan #66) builds on.++Pure; no database. Runs inside hspec via 'hspec-hedgehog'.+-}+module Test.Category (spec) where++import Data.Text (Text)+import Data.Text qualified as T+import Hedgehog+import Hedgehog.Gen qualified as Gen+import Hedgehog.Range qualified as Range+import Kiroku.Store.Types (CategoryName (..), StreamName (..), categoryName, streamNameInCategory)+import Test.Hspec+import Test.Hspec.Hedgehog (hedgehog)++{- | A non-empty, dash-free category. Includes @:@ and @$@ to exercise the+sub-namespacing convention (e.g. @"wf:orders"@) without the @-@ boundary.+-}+genCategoryText :: Gen Text+genCategoryText = Gen.text (Range.linear 1 16) (Gen.element categoryChars)+  where+    categoryChars = ['a' .. 'z'] <> ['0' .. '9'] <> [':', '$', '_']++{- | An arbitrary id segment. May contain @-@: the id lives after the first+@-@, so embedded dashes must not change the leading category.+-}+genSegment :: Gen Text+genSegment = Gen.text (Range.linear 0 16) (Gen.element segmentChars)+  where+    segmentChars = ['a' .. 'z'] <> ['0' .. '9'] <> ['-', ':', '_']++-- | An arbitrary stream name (may contain dashes anywhere, or none).+genStreamNameText :: Gen Text+genStreamNameText = Gen.text (Range.linear 0 24) (Gen.element nameChars)+  where+    nameChars = ['a' .. 'z'] <> ['0' .. '9'] <> ['-', ':', '_']++spec :: Spec+spec = describe "Kiroku.Store.Types category rule" $ do+    describe "categoryName" $ do+        it "takes the substring before the first dash" $+            categoryName (StreamName "orders-1") `shouldBe` CategoryName "orders"++        it "treats a dashless name as its own category" $+            categoryName (StreamName "singleton") `shouldBe` CategoryName "singleton"++        it "stops at the first dash even with later dashes" $+            categoryName (StreamName "orders-a-b-c") `shouldBe` CategoryName "orders"++        it "never yields a category containing a dash" $ hedgehog $ do+            name <- forAll genStreamNameText+            let CategoryName cat = categoryName (StreamName name)+            assert (not (T.isInfixOf "-" cat))++    describe "streamNameInCategory" $ do+        it "joins category and segment with a dash" $+            streamNameInCategory (CategoryName "orders") "1" `shouldBe` StreamName "orders-1"++    describe "round-trip" $ do+        it "categoryName . streamNameInCategory == id, for dash-free categories" $ hedgehog $ do+            cat <- forAll genCategoryText+            seg <- forAll genSegment+            categoryName (streamNameInCategory (CategoryName cat) seg) === CategoryName cat

@@ -249,6 +249,29 @@                         (V.toList allEvts)             sort positions `shouldBe` [1, 2, 3, 4, 5, 6] +    -- EP-4 M3 — Best-effort regression for the single-stream append retry.+    -- The race can pass vacuously on fast machines, but it must never surface+    -- PostgreSQL's transient transaction SQLSTATEs (40001/40P01) to callers as+    -- UnexpectedServerError.+    it "single-stream append retry does not leak transient SQLSTATEs" $+        withTestStore $ \store -> do+            forM_ [1 .. 25 :: Int] $ \i -> do+                let a = StreamName ("retry-a-" <> T.pack (show i))+                    b = StreamName ("retry-b-" <> T.pack (show i))+                    multi =+                        appendMultiStream+                            [ (a, NoStream, [makeEvent "A" (Aeson.object [])])+                            , (b, AnyVersion, [makeEvent "B-multi" (Aeson.object [])])+                            ]+                    single =+                        appendToStream b NoStream [makeEvent "B-single" (Aeson.object [])]+                (multiResult, singleResult) <-+                    Async.concurrently+                        (runStoreIO store multi)+                        (runStoreIO store single)+                assertNoTransientLeak ("multi round " <> show i) multiResult+                assertNoTransientLeak ("single round " <> show i) singleResult+ assertRightAppend :: Either StoreError AppendResult -> IO () assertRightAppend = \case     Right _ -> pure ()@@ -258,6 +281,14 @@ assertRightMulti = \case     Right _ -> pure ()     Left err -> expectationFailure ("appendMultiStream should succeed, got: " <> show err)++assertNoTransientLeak :: String -> Either StoreError a -> IO ()+assertNoTransientLeak label = \case+    Left (UnexpectedServerError code _)+        | code == "40P01" || code == "40001" ->+            expectationFailure (label <> ": transient SQLSTATE leaked as UnexpectedServerError " <> show code)+    _ ->+        pure ()  streamVersions :: V.Vector RecordedEvent -> [Int64] streamVersions =

@@ -28,6 +28,8 @@      -- * Raw SQL helpers     countEvents,+    countDeadLettersForEvents,+    insertDeadLetterForEvent,     insertEventUsingDefaultId,     serverVersionNum,     truncateRejected,@@ -53,9 +55,11 @@ import Control.Exception (SomeException) import Control.Lens ((^.)) import Data.Aeson (Value)+import Data.Aeson qualified as Aeson import Data.Generics.Labels () import Data.Int (Int32, Int64) import Data.Text (Text)+import Data.UUID (UUID) import Hasql.Connection qualified as Connection import Hasql.Connection.Settings qualified as Conn import Hasql.Decoders qualified as D@@ -64,6 +68,7 @@ import Hasql.Session qualified as Session import Hasql.Statement (Statement, preparable, unpreparable) import Kiroku.Store+import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.EventPublisher (publisherPosition) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) @@ -121,6 +126,41 @@     result <- Pool.use (store ^. #pool) (Session.statement () stmt)     case result of         Left err -> error ("countEvents failed: " <> show err)+        Right n -> pure n++-- | Insert a dead-letter row for a recorded event without running a subscription.+insertDeadLetterForEvent :: KirokuStore -> Text -> RecordedEvent -> IO ()+insertDeadLetterForEvent store subscriptionName event = do+    let EventId eid = event ^. #eventId+        GlobalPosition globalPosition = event ^. #globalPosition+        params =+            SQL.DeadLetterParams+                { SQL.dlSubscriptionName = subscriptionName+                , SQL.dlMember = 0+                , SQL.dlGlobalPosition = globalPosition+                , SQL.dlEventId = eid+                , SQL.dlReason = Aeson.object [("source", Aeson.String "test")]+                , SQL.dlReasonSummary = "test dead letter"+                , SQL.dlAttemptCount = 1+                }+    result <- Pool.use (store ^. #pool) (Session.statement params SQL.insertDeadLetterAndCheckpointStmt)+    case result of+        Left err -> error ("insertDeadLetterForEvent failed: " <> show err)+        Right () -> pure ()++-- | Count dead-letter rows referencing any of the supplied events.+countDeadLettersForEvents :: KirokuStore -> [EventId] -> IO Int64+countDeadLettersForEvents store eventIds = do+    let uuids = fmap (\(EventId eid) -> eid) eventIds+        stmt :: Statement [UUID] Int64+        stmt =+            preparable+                "SELECT COUNT(*) FROM dead_letters WHERE event_id = ANY($1::uuid[])"+                (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+                (D.singleRow (D.column (D.nonNullable D.int8)))+    result <- Pool.use (store ^. #pool) (Session.statement uuids stmt)+    case result of+        Left err -> error ("countDeadLettersForEvents failed: " <> show err)         Right n -> pure n  {- | Insert directly into @events@ without an @event_id@ so the database

@@ -0,0 +1,122 @@+module Test.NotifyGuard (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM (TVar, atomically, check, modifyTVar', newTVarIO, readTVar)+import Control.Exception (bracket, throwIO)+import Control.Lens ((^.))+import Data.Aeson qualified as Aeson+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.Generics.Labels ()+import Data.Text qualified as T+import Data.Text.Encoding qualified as T+import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings qualified as Conn+import Hasql.Notifications qualified as Notifications+import Kiroku.Store+import Kiroku.Test.Postgres (withMigratedTestDatabase)+import System.IO.Error (userError)+import Test.Helpers (makeEvent)+import Test.Hspec++spec :: Spec+spec =+    describe "NOTIFY trigger guard" $+        around withNotifyStore $ do+            it "emits one unchanged append payload and no lifecycle or $all payloads" $ \(store, listenerConn) -> do+                payloads <- newTVarIO []++                Notifications.listen listenerConn (Notifications.toPgIdentifier "kiroku.events")+                Async.withAsync+                    (Notifications.waitForNotifications (\_ payload -> atomically (modifyTVar' payloads (<> [payload]))) listenerConn)+                    $ \_listener -> do+                        let firstStream = StreamName "notify-guard-1"+                            secondStream = StreamName "notify-guard-2"++                        Right first <-+                            runStoreIO store $+                                appendToStream firstStream NoStream [makeEvent "NotifyGuardFirst" (Aeson.object [])]+                        waitForPayload payloads (expectedPayload firstStream first)++                        Right _ <- runStoreIO store $ softDeleteStream firstStream+                        Right _ <- runStoreIO store $ undeleteStream firstStream++                        Right second <-+                            runStoreIO store $+                                appendToStream secondStream NoStream [makeEvent "NotifyGuardSecond" (Aeson.object [])]+                        waitForPayload payloads (expectedPayload secondStream second)++                        actual <- atomically (readTVar payloads)+                        actual+                            `shouldBe` [ expectedPayload firstStream first+                                       , expectedPayload secondStream second+                                       ]+                        map payloadFieldCount actual `shouldBe` [3, 3]+                        actual `shouldSatisfy` all (not . BS.isPrefixOf "$all,")++            it "emits the unchanged payload for a 512-byte stream name" $ \(store, listenerConn) -> do+                payloads <- newTVarIO []++                Notifications.listen listenerConn (Notifications.toPgIdentifier "kiroku.events")+                Async.withAsync+                    (Notifications.waitForNotifications (\_ payload -> atomically (modifyTVar' payloads (<> [payload]))) listenerConn)+                    $ \_listener -> do+                        let stream = StreamName (T.replicate 512 "n")++                        Right result <-+                            runStoreIO store $+                                appendToStream stream NoStream [makeEvent "NotifyGuardMaxName" (Aeson.object [])]+                        let expected = expectedPayload stream result+                        waitForPayload payloads expected++                        actual <- atomically (readTVar payloads)+                        actual `shouldBe` [expected]+                        payloadFieldCount expected `shouldBe` 3++withNotifyStore :: ((KirokuStore, Connection.Connection) -> IO ()) -> IO ()+withNotifyStore action =+    withMigratedTestDatabase $ \connStr ->+        withStore (defaultConnectionSettings connStr) $ \store ->+            withConnection connStr $ \listenerConn ->+                action (store, listenerConn)++withConnection :: T.Text -> (Connection.Connection -> IO a) -> IO a+withConnection connStr =+    bracket acquire Connection.release+  where+    acquire = do+        result <- Connection.acquire (Conn.connectionString connStr)+        case result of+            Left err -> throwIO (userError ("failed to acquire LISTEN connection: " <> show err))+            Right conn -> pure conn++expectedPayload :: StreamName -> AppendResult -> ByteString+expectedPayload (StreamName name) result =+    let StreamId sid = result ^. #streamId+        StreamVersion version = result ^. #streamVersion+     in T.encodeUtf8 $+            T.intercalate+                ","+                [ name+                , T.pack (show sid)+                , T.pack (show version)+                ]++payloadFieldCount :: ByteString -> Int+payloadFieldCount = length . BS.split 44++waitForPayload :: TVar [ByteString] -> ByteString -> IO ()+waitForPayload payloads expected = do+    result <-+        Async.race+            (threadDelay 5_000_000)+            ( atomically $ do+                seen <- readTVar payloads+                check (expected `elem` seen)+            )+    case result of+        Left () -> do+            seen <- atomically (readTVar payloads)+            expectationFailure ("timed out waiting for notification " <> show expected <> "; saw " <> show seen)+        Right () -> pure ()

@@ -0,0 +1,106 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings #-}++module Test.PublisherCallbackResilience (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar)+import Control.Concurrent.STM (atomically, check, newTVarIO, readTVar, writeTVar)+import Control.Exception (Exception, throwIO)+import Control.Lens ((&), (.~), (^.))+import Data.Aeson qualified as Aeson+import Data.Data (Typeable)+import Data.Generics.Labels ()+import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef)+import Kiroku.Store+import Test.Helpers (caughtUpEventHandler, makeEvent, waitForSubscriptionLive, waitWithTimeout, withTestStoreSettings)+import Test.Hspec++data CallbackBoom = CallbackBoom+    deriving stock (Show, Typeable)+    deriving anyclass (Exception)++timeoutMicros :: Int+timeoutMicros = 10_000_000++within :: String -> IO a -> IO a+within label action = do+    result <- Async.race (threadDelay timeoutMicros) action+    case result of+        Left () -> fail ("timed out waiting for " <> label)+        Right a -> pure a++spec :: Spec+spec = describe "publisher callback resilience" $ do+    it "keeps the publisher alive when decodeHook throws once" $ do+        failedOnce <- newIORef False+        loopErrorSeen <- newEmptyMVar+        caughtUp <- newEmptyMVar+        delivered <- newIORef ([] :: [EventType])+        deliveredCount <- newTVarIO (0 :: Int)++        let subName = SubscriptionName "publisher-decode-hook-resilience"+            decodeOnce event+                | event ^. #eventType == EventType "Boom" = do+                    alreadyFailed <- atomicModifyIORef' failedOnce (\old -> (True, old))+                    if alreadyFailed+                        then pure event+                        else throwIO CallbackBoom+                | otherwise = pure event+            observe evt = do+                case evt of+                    KirokuEventPublisherLoopError{} -> () <$ tryPutMVar loopErrorSeen ()+                    _ -> pure ()+                caughtUpEventHandler subName caughtUp Nothing evt+            tweak settings =+                settings+                    & #storeSettings .~ defaultStoreSettings{decodeHook = Just decodeOnce}+                    & #eventHandler .~ Just observe+            handler event = do+                modifyIORef' delivered ((event ^. #eventType) :)+                atomically $ do+                    n <- readTVar deliveredCount+                    let n' = n + 1+                    writeTVar deliveredCount n'+                    pure $ if n' >= 2 then Stop else Continue++        withTestStoreSettings tweak $ \store -> do+            handle <- subscribe store (defaultSubscriptionConfig subName AllStreams handler)+            waitForSubscriptionLive caughtUp+            Right _ <- runStoreIO store $ appendToStream (StreamName "publisher-decode-boom") NoStream [makeEvent "Boom" (Aeson.object [])]+            within "publisher loop error event" (waitForSubscriptionLive loopErrorSeen)+            Right _ <- runStoreIO store $ appendToStream (StreamName "publisher-decode-ok") NoStream [makeEvent "Ok" (Aeson.object [])]+            result <- waitWithTimeout timeoutMicros handle+            case result of+                Right (Right ()) -> pure ()+                Left timeout -> expectationFailure timeout+                Right (Left e) -> expectationFailure ("expected clean stop, got: " <> show e)++        seen <- reverse <$> readIORef delivered+        seen `shouldBe` [EventType "Boom", EventType "Ok"]++    it "drops throwing eventHandler exceptions without killing publisher or worker" $ do+        deliveredCount <- newTVarIO (0 :: Int)+        let tweak settings = settings & #eventHandler .~ Just (\_ -> throwIO CallbackBoom)+            handler _ = do+                atomically $ do+                    n <- readTVar deliveredCount+                    let n' = n + 1+                    writeTVar deliveredCount n'+                    pure $ if n' >= 3 then Stop else Continue++        withTestStoreSettings tweak $ \store -> do+            handle <- subscribe store (defaultSubscriptionConfig (SubscriptionName "throwing-event-handler-resilience") AllStreams handler)+            Right _ <- runStoreIO store $ appendToStream (StreamName "throwing-handler-1") NoStream [makeEvent "One" (Aeson.object [])]+            Right _ <- runStoreIO store $ appendToStream (StreamName "throwing-handler-2") NoStream [makeEvent "Two" (Aeson.object [])]+            Right _ <- runStoreIO store $ appendToStream (StreamName "throwing-handler-3") NoStream [makeEvent "Three" (Aeson.object [])]+            result <- waitWithTimeout timeoutMicros handle+            case result of+                Right (Right ()) -> pure ()+                Left timeout -> expectationFailure timeout+                Right (Left e) -> expectationFailure ("expected clean stop, got: " <> show e)++        finalCount <- atomically (readTVar deliveredCount)+        finalCount `shouldBe` 3

@@ -0,0 +1,140 @@+{-# LANGUAGE OverloadedStrings #-}++module Test.PublisherIdleAdvance (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar)+import Control.Concurrent.STM (readTVarIO)+import Control.Exception (bracket)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (when)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.IntMap.Strict qualified as IntMap+import Data.List (sort)+import Data.String (fromString)+import Kiroku.Store+import Kiroku.Store.Subscription.EventPublisher qualified as Pub+import Test.Helpers (caughtUpEventHandler, makeEvent, waitForPublisher, waitForSubscriptionLive, withTestStoreSettings)+import Test.Hspec++timeoutMicros :: Int+timeoutMicros = 10_000_000++within :: String -> IO a -> IO a+within label action = do+    result <- Async.race (threadDelay timeoutMicros) action+    case result of+        Left () -> fail ("timed out waiting for " <> label)+        Right a -> pure a++countingSettings :: IORef Int -> ConnectionSettings -> ConnectionSettings+countingSettings counter settings =+    settings+        & #storeSettings+            .~ defaultStoreSettings+                { decodeHook =+                    Just $ \event -> do+                        atomicModifyIORef' counter (\n -> (n + 1, ()))+                        pure event+                }++appendEvents :: KirokuStore -> Int -> String -> IO GlobalPosition+appendEvents store n prefix = do+    results <-+        traverse+            ( \i ->+                runStoreIO store $+                    appendToStream+                        (StreamName (fromString (prefix <> "-" <> show i)))+                        NoStream+                        [makeEvent (fromString ("E" <> show i)) (Aeson.object [])]+            )+            [1 .. n]+    case sequence results of+        Left err -> fail ("append failed: " <> show err)+        Right [] -> fail "appendEvents called with zero events"+        Right xs -> pure (last xs ^. #globalPosition)++publisherSubscriberCount :: KirokuStore -> IO Int+publisherSubscriberCount store =+    IntMap.size <$> readTVarIO (Pub.subscribers (store ^. #publisher))++spec :: Spec+spec = describe "publisher idle advance" $ do+    it "advances lastPublished without decoding any rows when no subscriber is registered" $ do+        counter <- newIORef 0+        withTestStoreSettings (countingSettings counter) $ \store -> do+            tailPos <- appendEvents store 25 "pubidle-empty"+            waitForPublisher store tailPos+            readIORef counter `shouldReturn` 0++    it "does not fetch full rows while only a category subscriber exists" $ do+        counter <- newIORef 0+        caughtUp <- newEmptyMVar+        delivered <- newEmptyMVar+        let subName = SubscriptionName "pubidle-category"+            observe = caughtUpEventHandler subName caughtUp Nothing+            handler event = do+                case event ^. #eventType of+                    EventType "Wake" -> () <$ tryPutMVar delivered ()+                    _ -> pure ()+                pure Continue+            tweak settings =+                countingSettings counter settings+                    & #eventHandler .~ Just observe+        withTestStoreSettings tweak $ \store ->+            bracket+                (subscribe store (defaultSubscriptionConfig subName (Category (CategoryName "pubidlea")) handler))+                cancel+                $ \_handle -> do+                    within "category subscription live" (waitForSubscriptionLive caughtUp)+                    before <- readIORef counter+                    tailPos <- appendEvents store 30 "pubidleb"+                    waitForPublisher store tailPos+                    after <- readIORef counter+                    (after - before) `shouldBe` 0+                    publisherSubscriberCount store `shouldReturn` 0++                    Right wakePos <-+                        runStoreIO store $+                            appendToStream+                                (StreamName "pubidlea-1")+                                NoStream+                                [makeEvent "Wake" (Aeson.object [])]+                    waitForPublisher store (wakePos ^. #globalPosition)+                    within "category wake event" (waitForSubscriptionLive delivered)++    it "switches from cheap advance to all-stream delivery without gaps" $ do+        counter <- newIORef 0+        caughtUp <- newEmptyMVar+        seenRef <- newIORef ([] :: [GlobalPosition])+        seenEnough <- newEmptyMVar+        let subName = SubscriptionName "pubidle-transition"+            observe = caughtUpEventHandler subName caughtUp Nothing+            handler event = do+                let pos = event ^. #globalPosition+                count <-+                    atomicModifyIORef' seenRef $ \old ->+                        let new = pos : old+                         in (new, length new)+                when (count >= 35) (() <$ tryPutMVar seenEnough ())+                pure Continue+            tweak settings =+                countingSettings counter settings+                    & #eventHandler .~ Just observe+        withTestStoreSettings tweak $ \store -> do+            tailBefore <- appendEvents store 20 "pubidle-before"+            waitForPublisher store tailBefore+            bracket+                (subscribe store (defaultSubscriptionConfig subName AllStreams handler))+                cancel+                $ \_handle -> do+                    within "all-stream subscription live" (waitForSubscriptionLive caughtUp)+                    tailAfter <- appendEvents store 15 "pubidle-after"+                    waitForPublisher store tailAfter+                    within "all-stream events" (waitForSubscriptionLive seenEnough)+                    seen <- sort <$> readIORef seenRef+                    seen `shouldBe` fmap GlobalPosition [1 .. 35]

@@ -9,86 +9,145 @@ import Control.Lens ((^.)) import Data.Aeson qualified as Aeson import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef) import Data.List.NonEmpty qualified as NE import Data.Text qualified as T+import Data.UUID.V7 qualified as V7 import Data.Vector qualified as V import Kiroku.Store import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream-import Test.Helpers (makeEvent, withTestStore)+import Test.Helpers (makeEvent, waitForPublisher, withTestStore, withTestStoreSettings) import Test.Hspec  spec :: Spec-spec = around withTestStore $+spec = do     describe "readStreamForwardStream" $ do-        it "matches readStreamForward on a single-page stream" $ \store -> do-            let name = StreamName "rsfs-single-page"-                events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]-            Right _ <- runStoreIO store $ appendToStream name NoStream events-            streamed <- runStoreIO store $ Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)-            vectored <- runStoreIO store $ readStreamForward name (StreamVersion 0) 256-            case (streamed, vectored) of-                (Right xs, Right v) -> xs `shouldBe` V.toList v-                _ -> expectationFailure ("Unexpected error: " <> show streamed <> " / " <> show vectored)+        around withTestStore $ do+            it "matches readStreamForward on a single-page stream" $ \store -> do+                let name = StreamName "rsfs-single-page"+                    events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]+                Right _ <- runStoreIO store $ appendToStream name NoStream events+                streamed <- runStoreIO store $ Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)+                vectored <- runStoreIO store $ readStreamForward name (StreamVersion 0) 256+                case (streamed, vectored) of+                    (Right xs, Right v) -> xs `shouldBe` V.toList v+                    _ -> expectationFailure ("Unexpected error: " <> show streamed <> " / " <> show vectored) -        it "folds 1000 events end-to-end with Fold.length across multiple pages" $ \store -> do-            let name = StreamName "rsfs-multi-page"-                total = 1000 :: Int-                events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. total]]-            Right _ <- runStoreIO store $ appendToStream name NoStream events-            countResult <--                runStoreIO store $-                    Stream.fold Fold.length (readStreamForwardStream name (StreamVersion 0) 256)-            countResult `shouldBe` Right total-            allEvents <--                runStoreIO store $-                    Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)-            case allEvents of-                Right xs -> case NE.nonEmpty xs of-                    Just ne -> do-                        let h = NE.head ne-                            l = NE.last ne-                        (h ^. #streamVersion) `shouldBe` StreamVersion 1-                        (h ^. #eventType) `shouldBe` EventType (T.pack "E1")-                        (l ^. #streamVersion) `shouldBe` StreamVersion (fromIntegral total)-                        (l ^. #eventType) `shouldBe` EventType (T.pack ("E" <> show total))-                    Nothing -> expectationFailure "Expected non-empty events list"-                Left err -> expectationFailure ("Unexpected error: " <> show err)+            it "folds 1000 events end-to-end with Fold.length across multiple pages" $ \store -> do+                let name = StreamName "rsfs-multi-page"+                    total = 1000 :: Int+                    events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. total]]+                Right _ <- runStoreIO store $ appendToStream name NoStream events+                countResult <-+                    runStoreIO store $+                        Stream.fold Fold.length (readStreamForwardStream name (StreamVersion 0) 256)+                countResult `shouldBe` Right total+                allEvents <-+                    runStoreIO store $+                        Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)+                case allEvents of+                    Right xs -> case NE.nonEmpty xs of+                        Just ne -> do+                            let h = NE.head ne+                                l = NE.last ne+                            (h ^. #streamVersion) `shouldBe` StreamVersion 1+                            (h ^. #eventType) `shouldBe` EventType (T.pack "E1")+                            (l ^. #streamVersion) `shouldBe` StreamVersion (fromIntegral total)+                            (l ^. #eventType) `shouldBe` EventType (T.pack ("E" <> show total))+                        Nothing -> expectationFailure "Expected non-empty events list"+                    Left err -> expectationFailure ("Unexpected error: " <> show err) -        it "preserves order and avoids duplicates / gaps when paging at pageSize 2" $ \store -> do-            let name = StreamName "rsfs-page-boundary"-                events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]-            Right _ <- runStoreIO store $ appendToStream name NoStream events-            versions <--                runStoreIO store $-                    Stream.toList $-                        fmap (^. #streamVersion) (readStreamForwardStream name (StreamVersion 0) 2)-            versions-                `shouldBe` Right-                    [ StreamVersion 1-                    , StreamVersion 2-                    , StreamVersion 3-                    , StreamVersion 4-                    , StreamVersion 5-                    ]+            it "preserves order and avoids duplicates / gaps when paging at pageSize 2" $ \store -> do+                let name = StreamName "rsfs-page-boundary"+                    events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]+                Right _ <- runStoreIO store $ appendToStream name NoStream events+                versions <-+                    runStoreIO store $+                        Stream.toList $+                            fmap (^. #streamVersion) (readStreamForwardStream name (StreamVersion 0) 2)+                versions+                    `shouldBe` Right+                        [ StreamVersion 1+                        , StreamVersion 2+                        , StreamVersion 3+                        , StreamVersion 4+                        , StreamVersion 5+                        ] -        it "terminates immediately on a nonexistent stream" $ \store -> do-            let name = StreamName "rsfs-never-created"-            result <--                runStoreIO store $-                    Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)-            result `shouldBe` Right []+            it "terminates immediately on a nonexistent stream" $ \store -> do+                let name = StreamName "rsfs-never-created"+                result <-+                    runStoreIO store $+                        Stream.toList (readStreamForwardStream name (StreamVersion 0) 256)+                result `shouldBe` Right [] -        it "honors a non-zero starting cursor with exclusivity" $ \store -> do-            let name = StreamName "rsfs-non-zero-cursor"-                events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]-            Right _ <- runStoreIO store $ appendToStream name NoStream events-            tailEvents <--                runStoreIO store $-                    Stream.toList (readStreamForwardStream name (StreamVersion 2) 256)-            case tailEvents of-                Right xs -> do-                    length xs `shouldBe` 3-                    map (^. #streamVersion) xs-                        `shouldBe` [StreamVersion 3, StreamVersion 4, StreamVersion 5]-                Left err -> expectationFailure ("Unexpected error: " <> show err)+            it "honors a non-zero starting cursor with exclusivity" $ \store -> do+                let name = StreamName "rsfs-non-zero-cursor"+                    events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]+                Right _ <- runStoreIO store $ appendToStream name NoStream events+                tailEvents <-+                    runStoreIO store $+                        Stream.toList (readStreamForwardStream name (StreamVersion 2) 256)+                case tailEvents of+                    Right xs -> do+                        length xs `shouldBe` 3+                        map (^. #streamVersion) xs+                            `shouldBe` [StreamVersion 3, StreamVersion 4, StreamVersion 5]+                    Left err -> expectationFailure ("Unexpected error: " <> show err)++        it "uses one pool checkout per non-empty page when the final page is short" $ do+            ref <- newIORef (0 :: Int)+            let handler (ConnectionObservation _ InUseConnectionStatus) =+                    modifyIORef' ref (+ 1)+                handler _ =+                    pure ()+            withTestStoreSettings (\settings -> settings{observationHandler = Just handler}) $ \store -> do+                let name = StreamName "rsfs-short-page-count"+                    events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]+                Right appendResult <- runStoreIO store $ appendToStream name NoStream events+                waitForPublisher store (appendResult ^. #globalPosition)+                before <- readIORef ref+                streamed <- runStoreIO store $ Stream.toList (readStreamForwardStream name (StreamVersion 0) 2)+                after <- readIORef ref+                case streamed of+                    Right xs -> do+                        map (^. #streamVersion) xs+                            `shouldBe` [StreamVersion 1, StreamVersion 2, StreamVersion 3, StreamVersion 4, StreamVersion 5]+                        after - before `shouldBe` 3+                    Left err -> expectationFailure ("Unexpected error: " <> show err)++    describe "eventExistsInStream" $+        around withTestStore $ do+            it "finds an existing event id in its live stream" $ \store -> do+                eid <- EventId <$> V7.genUUID+                let name = StreamName "event-exists-present"+                Right _ <- runStoreIO store $ appendToStream name NoStream [eventWithId eid]+                runStoreIO store (eventExistsInStream name eid) `shouldReturn` Right True++            it "returns False for absent ids and ids linked only to another stream" $ \store -> do+                eid <- EventId <$> V7.genUUID+                otherEid <- EventId <$> V7.genUUID+                let source = StreamName "event-exists-source"+                    target = StreamName "event-exists-target"+                Right _ <- runStoreIO store $ appendToStream source NoStream [eventWithId eid]+                runStoreIO store (eventExistsInStream source otherEid) `shouldReturn` Right False+                runStoreIO store (eventExistsInStream target eid) `shouldReturn` Right False++            it "treats soft-deleted streams as absent" $ \store -> do+                eid <- EventId <$> V7.genUUID+                let name = StreamName "event-exists-soft-deleted"+                Right _ <- runStoreIO store $ appendToStream name NoStream [eventWithId eid]+                Right (Just _) <- runStoreIO store $ softDeleteStream name+                runStoreIO store (eventExistsInStream name eid) `shouldReturn` Right False++eventWithId :: EventId -> EventData+eventWithId eid =+    EventData+        { eventId = Just eid+        , eventType = EventType "Created"+        , payload = Aeson.object []+        , metadata = Nothing+        , causationId = Nothing+        , correlationId = Nothing+        }

@@ -0,0 +1,81 @@+{-# LANGUAGE OverloadedStrings #-}++module Test.StartupFailureSurfacing (spec) where++import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM (atomically, newTVarIO, readTVar, readTVarIO, writeTVar)+import Control.Exception (SomeException, fromException)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (replicateM_)+import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.IntMap.Strict qualified as IntMap+import Data.Map.Strict qualified as Map+import Hasql.Pool qualified as Pool+import Kiroku.Store+import Kiroku.Store.Subscription.EventPublisher qualified as Pub+import Kiroku.Store.Subscription.Worker (withLoadCheckpointHookForTest)+import Test.Helpers (waitWithTimeout, withTestStore, withTestStoreSettings)+import Test.Hspec++spec :: Spec+spec = describe "startup failure surfacing" $ do+    it "fails subscription startup loudly when checkpoint load fails" $ do+        handlerCalled <- newTVarIO False+        observedRef <- newIORef ([] :: [KirokuEvent])+        let subName = SubscriptionName "load-checkpoint-fails-loudly"+            cfg =+                defaultSubscriptionConfig subName AllStreams $ \_ -> do+                    atomically (writeTVar handlerCalled True)+                    pure Continue+            observe evt = modifyIORef' observedRef (evt :)+            injectLoadFailure _ = pure (Just (Left Pool.AcquisitionTimeoutUsageError))+            tweak settings = settings & #eventHandler .~ Just observe++        withTestStoreSettings tweak $ \store ->+            withLoadCheckpointHookForTest injectLoadFailure $ do+                handle <- subscribe store cfg+                result <- waitWithTimeout 5_000_000 handle+                case result of+                    Right (Left e)+                        | Just Pool.AcquisitionTimeoutUsageError <- fromException e -> pure ()+                    Left timeout -> expectationFailure timeout+                    Right other -> expectationFailure ("expected checkpoint load failure, got: " <> show other)++        wasCalled <- readTVarIO handlerCalled+        wasCalled `shouldBe` False+        observed <- reverse <$> readIORef observedRef+        any (isLoadCheckpointError subName) observed `shouldBe` True+        any (isCrashedStop subName) observed `shouldBe` True++    it "leaves no publisher or subscription registry entries after a subscribe/cancel storm" $+        withTestStore $ \store -> do+            let cfg = defaultSubscriptionConfig (SubscriptionName "subscribe-cancel-storm") AllStreams (\_ -> pure Continue)+            -- 200 iterations gives async exceptions many chances to land in the+            -- pre-fork window without making the test expensive on normal runs.+            replicateM_ 200 $ do+                pending <- Async.async (subscribe store cfg)+                Async.cancel pending+                outcome <- Async.waitCatch pending+                case outcome of+                    Right handle -> cancel handle >> (() <$ wait handle)+                    Left (_ :: SomeException) -> pure ()++            subs <- readTVarIO (Pub.subscribers (store ^. #publisher))+            reg <- readTVarIO (store ^. #subscriptionRegistry)+            IntMap.null subs `shouldBe` True+            Map.null reg `shouldBe` True++isLoadCheckpointError :: SubscriptionName -> KirokuEvent -> Bool+isLoadCheckpointError expected = \case+    KirokuEventSubscriptionDbError actual LoadCheckpoint Pool.AcquisitionTimeoutUsageError _+        | actual == expected -> True+    _ -> False++isCrashedStop :: SubscriptionName -> KirokuEvent -> Bool+isCrashedStop expected = \case+    KirokuEventSubscriptionStopped actual _ (StopWorkerCrashed e) _+        | actual == expected+        , Just Pool.AcquisitionTimeoutUsageError <- fromException e ->+            True+    _ -> False

@@ -0,0 +1,108 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings #-}++module Test.StreamBridgeTermination (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM (atomically, putTMVar)+import Control.Exception (Exception, SomeException, finally, fromException, throwIO, try)+import Control.Lens ((^.))+import Data.Aeson qualified as Aeson+import Data.Data (Typeable)+import Data.Generics.Labels ()+import Hasql.Pool qualified as Pool+import Kiroku.Store+import Kiroku.Store.Subscription.Stream (AckItem (..), InvalidStreamBufferSize (..), subscriptionAckStream)+import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest)+import Streamly.Data.Stream qualified as Stream+import Test.Helpers (makeEvent, waitForPublisher, withTestStore)+import Test.Hspec++data TestBoom = TestBoom+    deriving stock (Show, Typeable)+    deriving anyclass (Exception)++timeoutMicros :: Int+timeoutMicros = 10_000_000++within :: String -> IO a -> IO a+within label action = do+    result <- Async.race (threadDelay timeoutMicros) action+    case result of+        Left () -> fail ("timed out waiting for " <> label)+        Right a -> pure a++appendOne :: KirokuStore -> StreamName -> EventType -> IO GlobalPosition+appendOne store streamName (EventType eventType') = do+    Right r <- runStoreIO store $ appendToStream streamName NoStream [makeEvent eventType' (Aeson.object [])]+    pure (r ^. #globalPosition)++spec :: Spec+spec = describe "stream bridge termination" $ do+    it "rejects a zero-sized bridge buffer" $+        withTestStore $ \store -> do+            let cfg = defaultSubscriptionConfig (SubscriptionName "bridge-zero-buffer-sub") AllStreams (\_ -> pure Continue)+            subscriptionAckStream store cfg 0+                `shouldThrow` ( \e ->+                                    case e of+                                        InvalidStreamBufferSize 0 -> True+                                        _ -> False+                              )++    it "rethrows the worker exception to the consumer when the worker dies" $+        withTestStore $ \store -> do+            pos <- appendOne store (StreamName "bridge-crash") (EventType "BridgeCrash")+            waitForPublisher store pos+            let cfg = defaultSubscriptionConfig (SubscriptionName "bridge-crash-sub") AllStreams (\_ -> pure Continue)+                injectBoom _ _ = throwIO TestBoom+            withFetchBatchHookForTest injectBoom $ do+                (stream, cancelStream) <- subscriptionAckStream store cfg 16+                pulled <- within "stream pull to throw TestBoom" (try (Stream.uncons stream))+                cancelStream+                case pulled of+                    Left e+                        | Just TestBoom <- fromException (e :: SomeException) -> pure ()+                    Left e -> expectationFailure ("expected TestBoom, got: " <> show e)+                    Right _ -> expectationFailure "expected stream pull to throw TestBoom"++    it "ends the stream after a clean worker stop" $+        withTestStore $ \store -> do+            let cfg = defaultSubscriptionConfig (SubscriptionName "bridge-clean-stop-sub") AllStreams (\_ -> pure Continue)+            (stream0, cancelStream) <- subscriptionAckStream store cfg 16+            finally+                ( do+                    pos <- appendOne store (StreamName "bridge-clean-stop") (EventType "BridgeCleanStop")+                    waitForPublisher store pos+                    mFirst <- within "first bridge item" (Stream.uncons stream0)+                    (stream1, item) <- case mFirst of+                        Nothing -> fail "expected one bridge item before clean stop"+                        Just (firstItem, rest) -> pure (rest, firstItem)+                    atomically (putTMVar (ackReply item) Stop)+                    mNext <- within "stream end after clean stop" (Stream.uncons stream1)+                    case mNext of+                        Nothing -> pure ()+                        Just _ -> expectationFailure "expected stream to end after clean stop"+                )+                cancelStream++    it "cancelAction returns promptly even when the bridge queue is full" $+        withTestStore $ \store -> do+            let cfg = defaultSubscriptionConfig (SubscriptionName "bridge-full-cancel-sub") AllStreams (\_ -> pure Continue)+            (stream0, cancelStream) <- subscriptionAckStream store cfg 1+            pos1 <- appendOne store (StreamName "bridge-full-cancel-1") (EventType "BridgeFullCancel1")+            pos2 <- appendOne store (StreamName "bridge-full-cancel-2") (EventType "BridgeFullCancel2")+            waitForPublisher store pos2++            within "non-blocking cancelAction with a full queue" cancelStream++            mFirst <- within "drained item after cancel" (Stream.uncons stream0)+            case mFirst of+                Nothing -> pure ()+                Just (item, stream1) -> do+                    atomically (putTMVar (ackReply item) Continue)+                    mNext <- within "stream end after cancel" (Stream.uncons stream1)+                    case mNext of+                        Nothing -> pure ()+                        Just _ -> expectationFailure "expected stream to end after cancel"

@@ -8,15 +8,16 @@ import Control.Lens ((^.)) import Data.Aeson qualified as Aeson import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Map.Strict qualified as Map import Data.Vector qualified as V import Kiroku.Store-import Test.Helpers (makeEvent, withTestStore)+import Test.Helpers (makeEvent, waitForPublisher, withTestStore, withTestStoreSettings) import Test.Hspec  spec :: Spec-spec = around withTestStore $-    describe "lookupStreamNames / lookupStreamName" $ do+spec = describe "lookupStreamNames / lookupStreamName" $ do+    around withTestStore $ do         it "resolves originalStreamId back to the source stream for $all reads" $ \store -> do             Right _ <-                 runStoreIO store $@@ -50,3 +51,28 @@             found `shouldBe` Just (StreamName "widgets-7")             Right missing <- runStoreIO store $ lookupStreamName (StreamId 888888)             missing `shouldBe` Nothing++    it "short-circuits empty input without a pool checkout" $ do+        ref <- newIORef (0 :: Int)+        let handler (ConnectionObservation _ InUseConnectionStatus) =+                modifyIORef' ref (+ 1)+            handler _ =+                pure ()+        withTestStoreSettings (\settings -> settings{observationHandler = Just handler}) $ \store -> do+            Right appendResult <-+                runStoreIO store $+                    appendToStream (StreamName "lookup-count-1") NoStream [makeEvent "LookupCounted" (Aeson.object [])]+            waitForPublisher store (appendResult ^. #globalPosition)+            Right (Just sid) <- runStoreIO store $ lookupStreamId (StreamName "lookup-count-1")++            beforeEmpty <- readIORef ref+            Right emptyNames <- runStoreIO store $ lookupStreamNames []+            afterEmpty <- readIORef ref+            emptyNames `shouldBe` Map.empty+            afterEmpty - beforeEmpty `shouldBe` 0++            beforeReal <- readIORef ref+            Right names <- runStoreIO store $ lookupStreamNames [sid]+            afterReal <- readIORef ref+            names `shouldBe` Map.singleton sid (StreamName "lookup-count-1")+            afterReal - beforeReal `shouldBe` 1

@@ -21,16 +21,19 @@ module Test.SubscriptionRegistry (spec) where  import Control.Concurrent (threadDelay)+import Control.Concurrent.STM (readTVarIO) import Control.Exception (throwIO) import Control.Lens ((^.)) import Data.Aeson qualified as Aeson import Data.Foldable (for_) import Data.Generics.Labels () import Data.Int (Int32)+import Data.IntMap.Strict qualified as IntMap import Data.Map.Strict qualified as Map import Data.Maybe (isJust, isNothing) import Data.Text (Text) import Kiroku.Store+import Kiroku.Store.Subscription.EventPublisher qualified as Pub import Kiroku.Store.Subscription.Types (SubscriptionConfigM (..)) import Test.Helpers (makeEvent, waitForPublisher, withTestStore) import Test.Hspec@@ -46,6 +49,17 @@         { consumerGroup = Just (ConsumerGroup{member = m, size = n})         } +-- | A size-@n@ @$all@ group config for member @m@ whose handler never stops.+groupAllCont :: Text -> Int32 -> Int32 -> SubscriptionConfig+groupAllCont nm m n =+    (defaultSubscriptionConfig (SubscriptionName nm) AllStreams (\_ -> pure Continue))+        { consumerGroup = Just (ConsumerGroup{member = m, size = n})+        }++publisherSubscriberCount :: KirokuStore -> IO Int+publisherSubscriberCount store =+    IntMap.size <$> readTVarIO (Pub.subscribers (store ^. #publisher))+ -- | Poll the snapshot until the keyed entry reaches the given 'statePhase' label. waitUntilPhase :: Int -> KirokuStore -> (SubscriptionName, Int32) -> Text -> IO Bool waitUntilPhase budget store key phase@@ -168,3 +182,37 @@             cs2 <- currentState h2             cs2 `shouldSatisfy` isJust             cancel h2++    describe "publisher queue registration" $ do+        it "registers publisher queues only for non-group AllStreams subscriptions" $+            withTestStore $ \store -> do+                initial <- publisherSubscriberCount store+                initial `shouldBe` 0++                hCategory <- subscribe store (defaultSubscriptionConfig (SubscriptionName "reg-cat-no-queue") (Category (CategoryName "regcat")) (\_ -> pure Continue))+                publisherSubscriberCount store `shouldReturn` 0++                hGroupCategory <- subscribe store (groupCont "reg-group-cat-no-queue" "regcat" 0 2)+                publisherSubscriberCount store `shouldReturn` 0++                hGroupAll <- subscribe store (groupAllCont "reg-group-all-no-queue" 0 2)+                publisherSubscriberCount store `shouldReturn` 0++                hAll <- subscribe store (plainCont "reg-all-has-queue")+                publisherSubscriberCount store `shouldReturn` 1++                cancel hAll+                _ <- wait hAll+                publisherSubscriberCount store `shouldReturn` 0++                cancel hCategory+                _ <- wait hCategory+                publisherSubscriberCount store `shouldReturn` 0++                cancel hGroupCategory+                _ <- wait hGroupCategory+                publisherSubscriberCount store `shouldReturn` 0++                cancel hGroupAll+                _ <- wait hGroupAll+                publisherSubscriberCount store `shouldReturn` 0

@@ -12,7 +12,9 @@ import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Int (Int64) import Data.Text (Text)+import Data.Text qualified as T import Data.Time.Clock (getCurrentTime)+import Data.UUID.V7 qualified as V7 import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E import Hasql.Pool (Pool)@@ -109,6 +111,23 @@             rows <- countSideTable (store ^. #pool)             rows `shouldBe` 0 +        it "surfaces duplicate caller-supplied event ids as DuplicateEvent" $ \store -> do+            eid <- EventId <$> V7.genUUID+            let evt = eventWithId eid+            prepared <- prepareEventsIO [evt]+            now <- getCurrentTime+            first <-+                runStoreIO store $+                    runTransaction $+                        appendToStreamTx (StreamName "txn-duplicate-source-1") NoStream prepared now+            first `shouldSatisfy` either (const False) isRight++            duplicate <-+                runStoreIO store $+                    runTransaction $+                        appendToStreamTx (StreamName "txn-duplicate-target-1") NoStream prepared now+            duplicate `shouldSatisfy` isDuplicateEvent+     describe "runTransactionAppending" $ do         it "commits 3 events and a side-table row in one ACID transaction" $ \store -> do             createSideTable (store ^. #pool)@@ -226,6 +245,64 @@             rows <- countSideTable (store ^. #pool)             rows `shouldBe` 0 +        it "rejects oversized stream names before opening any transaction" $ \store -> do+            createSideTable (store ^. #pool)+            let over = StreamName (T.replicate 513 "t")+            result <-+                runStoreIO store $+                    runTransactionAppending+                        over+                        AnyVersion+                        [makeEvent "Should" (Aeson.object [])]+                        ( \ar -> do+                            let StreamId sid = ar ^. #streamId+                            Tx.statement (sid, "should-never-run") insertSideRowStmt+                            pure ar+                        )+            result `shouldBe` Right (Left (StreamNameTooLong over 513))+            runStoreIO store (getStream over) `shouldReturn` Right Nothing+            rows <- countSideTable (store ^. #pool)+            rows `shouldBe` 0++        it "rejects empty event batches before opening a transaction" $ \store -> do+            createSideTable (store ^. #pool)+            result <-+                runStoreIO store $+                    runTransactionAppending+                        (StreamName "txn-wrapper-empty-1")+                        NoStream+                        []+                        ( \ar -> do+                            let StreamId sid = ar ^. #streamId+                            Tx.statement (sid, "should-never-run") insertSideRowStmt+                            pure ar+                        )+            result `shouldBe` Right (Left (EmptyAppendBatch (StreamName "txn-wrapper-empty-1")))+            runStoreIO store (getStream (StreamName "txn-wrapper-empty-1")) `shouldReturn` Right Nothing+            rows <- countSideTable (store ^. #pool)+            rows `shouldBe` 0++        it "surfaces duplicate caller-supplied event ids as DuplicateEvent" $ \store -> do+            eid <- EventId <$> V7.genUUID+            let evt = eventWithId eid+            first <-+                runStoreIO store $+                    runTransactionAppending+                        (StreamName "txn-wrapper-duplicate-source-1")+                        NoStream+                        [evt]+                        pure+            first `shouldSatisfy` either (const False) isRight++            duplicate <-+                runStoreIO store $+                    runTransactionAppending+                        (StreamName "txn-wrapper-duplicate-target-1")+                        NoStream+                        [evt]+                        pure+            duplicate `shouldSatisfy` isDuplicateEvent+ -- --------------------------------------------------------------------------- -- Local statements and pool-side helpers -- ---------------------------------------------------------------------------@@ -283,3 +360,24 @@     case r of         Left err -> error ("countSideTable failed: " <> show err)         Right n -> pure n++isRight :: Either a b -> Bool+isRight = \case+    Right _ -> True+    Left _ -> False++isDuplicateEvent :: Either StoreError a -> Bool+isDuplicateEvent = \case+    Left (DuplicateEvent _) -> True+    _ -> False++eventWithId :: EventId -> EventData+eventWithId eid =+    EventData+        { eventId = Just eid+        , eventType = EventType "Created"+        , payload = Aeson.object []+        , metadata = Nothing+        , causationId = Nothing+        , correlationId = Nothing+        }

@@ -0,0 +1,125 @@+{- | Tests for the logical truncate-before marker (close-the-book+compaction, ExecPlan docs/plans/65).++The marker is a per-stream cursor that hides a prefix from the /ordered+per-stream/ reads ('readStreamForward' / 'readStreamBackward' and the paged+'readStreamForwardStream') while leaving the global @$all@ log, category+reads, and the physical row count untouched. The distinguishing property —+and the single most important assertion here — is that after a truncate the+@$all@ log and 'readCategory' still return the full history and+'countEvents' is unchanged: nothing is deleted, only hidden, and the+operation is fully reversible.+-}+module Test.TruncateBefore (spec) where++import Control.Lens ((^.))+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.Text qualified as T+import Data.Vector qualified as V+import Kiroku.Store+import Streamly.Data.Stream qualified as Stream+import Test.Helpers (countEvents, makeEvent, withTestStore)+import Test.Hspec++-- | Append @n@ events @E1 .. En@ to a fresh stream and return its name.+seedStream :: KirokuStore -> StreamName -> Int -> IO ()+seedStream store name n = do+    let events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. n]]+    Right _ <- runStoreIO store $ appendToStream name NoStream events+    pure ()++-- | The per-stream versions returned by a forward read, in order.+forwardVersions :: KirokuStore -> StreamName -> IO [StreamVersion]+forwardVersions store name = do+    Right v <- runStoreIO store $ readStreamForward name (StreamVersion 0) 100+    pure (map (^. #streamVersion) (V.toList v))++spec :: Spec+spec = describe "TruncateBefore" $+    around withTestStore $ do+        it "bounds per-stream reads to the kept suffix" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6 -- versions 1..6; treat v6 as the snapshot+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)++            forwardVersions store name `shouldReturn` [StreamVersion 6]++            Right back <- runStoreIO store $ readStreamBackward name (StreamVersion 0) 100+            map (^. #streamVersion) (V.toList back) `shouldBe` [StreamVersion 6]++        it "leaves the $all global log and category reads intact" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6+            before <- countEvents store+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)++            -- \$all global log still returns the full history.+            Right allEvents <- runStoreIO store $ readAllForward (GlobalPosition 0) 1000+            length (V.toList allEvents) `shouldBe` 6++            -- Category read still returns the full history.+            Right catEvents <- runStoreIO store $ readCategory (CategoryName "preference") (GlobalPosition 0) 1000+            length (V.toList catEvents) `shouldBe` 6++            -- Nothing was physically deleted.+            after <- countEvents store+            after `shouldBe` before++        it "is reversible via clearStreamTruncateBefore" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)+            forwardVersions store name `shouldReturn` [StreamVersion 6]++            Right (Just _) <- runStoreIO store $ clearStreamTruncateBefore name+            forwardVersions store name+                `shouldReturn` map StreamVersion [1 .. 6]++        it "is idempotent" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)+            first <- forwardVersions store name+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)+            second <- forwardVersions store name+            first `shouldBe` [StreamVersion 4, StreamVersion 5, StreamVersion 6]+            second `shouldBe` first++        it "applies across paged reads" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 4)+            -- Page size 2 forces multiple pages across the kept suffix.+            Right streamed <-+                runStoreIO store $+                    Stream.toList $+                        fmap (^. #streamVersion) (readStreamForwardStream name (StreamVersion 0) 2)+            streamed `shouldBe` [StreamVersion 4, StreamVersion 5, StreamVersion 6]++        it "reflects the marker in getStream" $ \store -> do+            let name = StreamName "preference-abc"+            seedStream store name 6+            Right (Just info0) <- runStoreIO store $ getStream name+            (info0 ^. #truncateBefore) `shouldBe` StreamVersion 0+            Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 3)+            Right (Just info1) <- runStoreIO store $ getStream name+            (info1 ^. #truncateBefore) `shouldBe` StreamVersion 3++        it "returns Nothing for missing or soft-deleted streams" $ \store -> do+            Right missing <- runStoreIO store $ setStreamTruncateBefore (StreamName "does-not-exist") (StreamVersion 1)+            missing `shouldBe` Nothing++            let name = StreamName "preference-abc"+            seedStream store name 3+            Right (Just _) <- runStoreIO store $ softDeleteStream name+            Right deleted <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 1)+            deleted `shouldBe` Nothing++        it "rejects $all with ReservedStreamName" $ \store -> do+            result <- runStoreIO store $ setStreamTruncateBefore (StreamName "$all") (StreamVersion 1)+            case result of+                Left (ReservedStreamName (StreamName "$all")) -> pure ()+                other ->+                    expectationFailure+                        ("Expected Left (ReservedStreamName \"$all\"), got: " <> show other)