kiroku-store 0.6.0.0 → 0.7.0.0

raw patch · 18 files changed

+1722/−21 lines, 18 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +34/−0
• kiroku-store.cabal +8/−1
• src/Kiroku/Store.hs +2/−0
• src/Kiroku/Store/Effect.hs +90/−14
• src/Kiroku/Store/Error.hs +5/−0
• src/Kiroku/Store/HistoryRetention.hs +147/−0
• src/Kiroku/Store/HistoryRetention/Internal.hs +91/−0
• src/Kiroku/Store/HistoryRetention/SQL.hs +298/−0
• src/Kiroku/Store/HistoryRetention/Types.hs +180/−0
• src/Kiroku/Store/Lifecycle.hs +15/−2
• src/Kiroku/Store/Observability.hs +21/−0
• src/Kiroku/Store/Transaction.hs +5/−0
• test/Main.hs +6/−0
• test/Test/HistoryRetention.hs +376/−0
• test/Test/HistoryRetentionMock.hs +106/−0
• test/Test/PerformanceStructure.hs +84/−1
• test/Test/StreamHistoryGuard.hs +205/−0
• test/Test/SubscriptionCheckpointInventory.hs +49/−3

@@ -1,5 +1,39 @@ # Changelog +## 0.7.0.0 — 2026-08-13++### Breaking Changes++* The exported `Store` effect gains acquire, renew, release, inventory, and+  prune operations for replay-history retention. Exhaustive custom and mock+  interpreters must handle the five new constructors.+* `KirokuEvent` gains committed retention acquisition, renewal, release,+  pruning, and hard-delete-conflict events. Exhaustive handlers must handle the+  new constructors.+* `StoreError` gains `HistoryRetentionActive`, returned when supported hard+  delete is refused by an active lease.++### New Features++* New `Kiroku.Store.HistoryRetention` and+  `Kiroku.Store.HistoryRetention.Types` modules expose validated, durable,+  database-time-derived leases for stable long rebuilds. Transaction+  combinators and mockable effect wrappers share the same owner-aware acquire,+  renew, release, bounded inventory, and terminal-row pruning semantics.+* `lockStreamHistoryForReplayTx` holds a transaction-scoped stream guard, and+  `readStreamForwardTx` reads exact ordered history in that same transaction.+  Append, link, lifecycle mutation, and every supported hard delete affecting+  the stream serialize behind the guard.+* Supported hard delete checks retention before mutation and locks every+  affected stream in ascending ID order, including streams containing links to+  target-originated events.++### Other Changes++* Added deterministic coordinator-race, raw `DELETE`/`TRUNCATE`, rollback,+  linked-history, deadlock, hot-path exclusion, query-plan, and controlled+  workload coverage for replay-history protection.+ ## 0.6.0.0 — 2026-08-12  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.6.0.0+version:         0.7.0.0 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -38,6 +38,8 @@     Kiroku.Store.Effect     Kiroku.Store.Effect.Resource     Kiroku.Store.Error+    Kiroku.Store.HistoryRetention+    Kiroku.Store.HistoryRetention.Types     Kiroku.Store.Lifecycle     Kiroku.Store.Link     Kiroku.Store.Notification@@ -57,6 +59,8 @@     Kiroku.Store.Types    other-modules:+    Kiroku.Store.HistoryRetention.Internal+    Kiroku.Store.HistoryRetention.SQL     Kiroku.Store.Subscription.Checkpoint.SQL     Kiroku.Store.Subscription.CheckpointInventory.SQL @@ -103,6 +107,8 @@     Test.EventTypeFilter     Test.FailureInjection     Test.Helpers+    Test.HistoryRetention+    Test.HistoryRetentionMock     Test.InterpreterHooks     Test.NotifyGuard     Test.PerformanceStructure@@ -113,6 +119,7 @@     Test.ReadStream     Test.StartupFailureSurfacing     Test.StreamBridgeTermination+    Test.StreamHistoryGuard     Test.StreamNameLookup     Test.SubscriptionCheckpointInitialization     Test.SubscriptionCheckpointInitializationMock

@@ -13,6 +13,7 @@     module Kiroku.Store.Effect,     module Kiroku.Store.Effect.Resource,     module Kiroku.Store.Error,+    module Kiroku.Store.HistoryRetention,     module Kiroku.Store.Append,     module Kiroku.Store.Causation,     module Kiroku.Store.Lifecycle,@@ -55,6 +56,7 @@ import Kiroku.Store.Effect import Kiroku.Store.Effect.Resource import Kiroku.Store.Error+import Kiroku.Store.HistoryRetention import Kiroku.Store.Lifecycle import Kiroku.Store.Link import Kiroku.Store.Notification (NotifierStartError (..))

@@ -23,7 +23,6 @@ import Control.Monad.Except qualified as Except import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value)-import Data.Foldable (for_) import Data.Generics.Labels () import Data.Int (Int32, Int64) import Data.List (find)@@ -53,7 +52,9 @@ import Kiroku.Store.Connection (KirokuStore (..)) import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore) import Kiroku.Store.Error (StoreError (..), attributeMultiStreamError, emptyResultError, isTransientSerializationError, mapLinkUsageError, mapTransactionUsageError, mapUsageError, validateStreamName)-import Kiroku.Store.Observability (KirokuEvent (..))+import Kiroku.Store.HistoryRetention.Internal qualified as HistoryRetention+import Kiroku.Store.HistoryRetention.Types+import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (decodeEvents, enrichEvents) import Kiroku.Store.Subscription.Checkpoint.SQL qualified as CheckpointSQL@@ -158,6 +159,18 @@         Int32 ->         MissingCheckpointPolicy ->         Store m (Either SubscriptionCheckpointMissing CheckpointInitialization)+    AcquireHistoryRetentionLease :: HistoryRetentionLeaseRequest -> Store m HistoryRetentionLease+    RenewHistoryRetentionLease ::+        HistoryRetentionLeaseHandle ->+        HistoryRetentionLeaseDuration ->+        Store m (Either HistoryRetentionRenewalError HistoryRetentionLease)+    ReleaseHistoryRetentionLease ::+        HistoryRetentionLeaseHandle ->+        Store m HistoryRetentionReleaseResult+    GetHistoryRetentionLeaseInventory ::+        HistoryRetentionInventoryQuery ->+        Store m (Vector HistoryRetentionLease)+    PruneHistoryRetentionLeases :: UTCTime -> Store m HistoryRetentionPruneResult     {- | 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@@ -346,18 +359,24 @@         rejectInvalidApplicationStream name         let txn = do                 Tx.sql "SET LOCAL kiroku.enable_hard_deletes = 'on'"+                HistoryRetention.lockHistoryRetentionCoordinatorTx                 mSid <- Tx.statement name SQL.findStreamIdStmt                 case mSid of-                    Nothing -> pure Nothing+                    Nothing -> pure (Right Nothing)                     Just sid -> do-                        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))+                        conflict <- HistoryRetention.activeHistoryRetentionConflictTx+                        case conflict of+                            Just active -> pure (Left active)+                            Nothing -> do+                                HistoryRetention.lockAffectedStreamsForHardDeleteTx sid+                                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 (Right (Just (StreamId sid)))         result <-             usePool (store ^. #pool) $                 TxSessions.transaction TxSessions.ReadCommitted TxSessions.Write txn@@ -366,9 +385,16 @@         -- application-level event before calling hardDeleteStream — see         -- docs/PRODUCTION-DEPLOYMENT.md.         case result of-            Just sid -> liftIO $ for_ (store ^. #eventHandler) ($ KirokuEventHardDeleteIssued (StreamName name) sid)-            Nothing -> pure ()-        pure result+            Right (Just sid) ->+                liftIO $ emitOrDrop (store ^. #eventHandler) (KirokuEventHardDeleteIssued (StreamName name) sid)+            Right Nothing -> pure ()+            Left conflict -> do+                liftIO $+                    emitOrDrop+                        (store ^. #eventHandler)+                        (KirokuEventHardDeleteHistoryRetentionConflict (StreamName name) conflict)+                throwError (HistoryRetentionActive (StreamName name) conflict)+        pure (either (const Nothing) (\value -> value) result)     UndeleteStream (StreamName name) -> do         rejectInvalidApplicationStream name         usePool (store ^. #pool) $@@ -383,6 +409,56 @@     InitializeSubscriptionCheckpoint subscriptionName member policy ->         usePool (store ^. #pool) $             CheckpointSQL.initializeSubscriptionCheckpointSession subscriptionName member policy+    AcquireHistoryRetentionLease request -> do+        lease <- runTxOnPool (store ^. #pool) TxSessions.transaction (HistoryRetention.acquireHistoryRetentionLeaseTx request)+        let HistoryRetentionLease{leaseId, owner, protectedThrough, expiresAt} = lease+        liftIO $+            emitOrDrop+                (store ^. #eventHandler)+                (KirokuEventHistoryRetentionLeaseAcquired leaseId owner protectedThrough expiresAt)+        pure lease+    RenewHistoryRetentionLease handle duration -> do+        result <-+            runTxOnPool+                (store ^. #pool)+                TxSessions.transaction+                (HistoryRetention.renewHistoryRetentionLeaseTx handle duration)+        case result of+            Right lease ->+                let HistoryRetentionLease{leaseId, owner, expiresAt} = lease+                 in liftIO $+                        emitOrDrop+                            (store ^. #eventHandler)+                            (KirokuEventHistoryRetentionLeaseRenewed leaseId owner expiresAt)+            Left _ -> pure ()+        pure result+    ReleaseHistoryRetentionLease handle -> do+        result <-+            runTxOnPool+                (store ^. #pool)+                TxSessions.transaction+                (HistoryRetention.releaseHistoryRetentionLeaseTx handle)+        case result of+            HistoryRetentionReleased HistoryRetentionLease{leaseId, owner, releasedAt = Just releasedAt} ->+                liftIO $+                    emitOrDrop+                        (store ^. #eventHandler)+                        (KirokuEventHistoryRetentionLeaseReleased leaseId owner releasedAt)+            _ -> pure ()+        pure result+    GetHistoryRetentionLeaseInventory query ->+        runTxOnPool+            (store ^. #pool)+            TxSessions.transaction+            (HistoryRetention.historyRetentionLeaseInventoryTx query)+    PruneHistoryRetentionLeases cutoff -> do+        result <-+            runTxOnPool+                (store ^. #pool)+                TxSessions.transaction+                (HistoryRetention.pruneHistoryRetentionLeasesTx cutoff)+        liftIO $ emitOrDrop (store ^. #eventHandler) (KirokuEventHistoryRetentionLeasesPruned result)+        pure result     RunTransaction tx ->         runTxOnPool (store ^. #pool) TxSessions.transaction tx     RunTransactionNoRetry tx ->

@@ -29,6 +29,7 @@ import GHC.Generics (Generic) import Hasql.Errors qualified as Errors import Hasql.Pool (UsageError (..))+import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict) import Kiroku.Store.Types  {- | Errors that can occur during store operations.@@ -69,6 +70,10 @@       EmptyAppendBatch !StreamName     | -- | The named stream does not exist (or has been soft-deleted).       StreamNotFound !StreamName+    | {- | A supported hard delete found one or more active replay-history+      retention leases and committed no destructive work.+      -}+      HistoryRetentionActive !StreamName !HistoryRetentionConflict     | {- | The named stream is reserved for store internals and cannot be       used as an application stream target. For now this applies only       to @$all@, which is the global read stream backed by the seeded

@@ -0,0 +1,147 @@+{- | Durable replay-history retention and transaction-scoped stream guards.++A lease protects the retained global event set from Kiroku hard delete and+authorized direct SQL deletion until its database-derived expiry. Transaction+combinators persist durable evidence but cannot emit process-local observability+events from inside an opaque caller-owned transaction.++Lease duration is one second through one hour. Expiry is passive: a crashed+owner blocks destructive work only until PostgreSQL time reaches @expiresAt@.+The owner is an accidental-mutation guard, not an authorization credential;+database roles and grants remain the security boundary. Direct destructive SQL+receives SQLSTATE @KR001@ while any lease is active.++When one transaction needs both a lease operation and a stream-history guard,+perform the lease operation first. This preserves Kiroku's coordinator-before-+stream-row lock order.+-}+module Kiroku.Store.HistoryRetention (+    module Kiroku.Store.HistoryRetention.Types,+    acquireHistoryRetentionLeaseTx,+    renewHistoryRetentionLeaseTx,+    releaseHistoryRetentionLeaseTx,+    historyRetentionLeaseInventoryTx,+    pruneHistoryRetentionLeasesTx,+    acquireHistoryRetentionLease,+    renewHistoryRetentionLease,+    releaseHistoryRetentionLease,+    historyRetentionLeaseInventory,+    pruneHistoryRetentionLeases,+    lockStreamHistoryForReplayTx,+    readStreamForwardTx,+) where++import Data.Int (Int32)+import Data.Time.Clock (UTCTime)+import Data.Vector (Vector)+import Effectful (Eff, (:>))+import Effectful.Dispatch.Dynamic (send)+import Hasql.Transaction qualified as Tx+import Kiroku.Store.Effect (Store (..))+import Kiroku.Store.HistoryRetention.Internal qualified as Internal+import Kiroku.Store.HistoryRetention.SQL qualified as HistoryRetentionSQL+import Kiroku.Store.HistoryRetention.Types+import Kiroku.Store.SQL qualified as SQL+import Kiroku.Store.Types (RecordedEvent, StreamInfo, StreamName (..), StreamVersion (..))++{- | Acquire a new lease and atomically capture the authoritative @$all@+frontier as the inclusive @protectedThrough@ rebuild ceiling. PostgreSQL+supplies every timestamp. Rollback leaves no lease row.+-}+acquireHistoryRetentionLeaseTx :: HistoryRetentionLeaseRequest -> Tx.Transaction HistoryRetentionLease+acquireHistoryRetentionLeaseTx = Internal.acquireHistoryRetentionLeaseTx++{- | Renew a still-active lease without shortening its current expiry.+Unknown, wrong-owner, expired, and released leases return distinct typed+errors and are never resurrected.+-}+renewHistoryRetentionLeaseTx ::+    HistoryRetentionLeaseHandle ->+    HistoryRetentionLeaseDuration ->+    Tx.Transaction (Either HistoryRetentionRenewalError HistoryRetentionLease)+renewHistoryRetentionLeaseTx = Internal.renewHistoryRetentionLeaseTx++-- | Release a live lease. Repetition is typed and does not rewrite the row.+releaseHistoryRetentionLeaseTx ::+    HistoryRetentionLeaseHandle ->+    Tx.Transaction HistoryRetentionReleaseResult+releaseHistoryRetentionLeaseTx = Internal.releaseHistoryRetentionLeaseTx++{- | Read a deterministic, database-time-derived, bounded inventory. Active,+expired, and released state is derived at statement time; no expiry worker is+required.+-}+historyRetentionLeaseInventoryTx ::+    HistoryRetentionInventoryQuery ->+    Tx.Transaction (Vector HistoryRetentionLease)+historyRetentionLeaseInventoryTx = Internal.historyRetentionLeaseInventoryTx++-- | Remove only expired or released rows strictly older than the supplied cutoff.+pruneHistoryRetentionLeasesTx :: UTCTime -> Tx.Transaction HistoryRetentionPruneResult+pruneHistoryRetentionLeasesTx = Internal.pruneHistoryRetentionLeasesTx++{- | Acquire a lease through the mockable 'Store' effect. The production+interpreter emits the committed acquisition event after the transaction ends.+-}+acquireHistoryRetentionLease :: (Store :> es) => HistoryRetentionLeaseRequest -> Eff es HistoryRetentionLease+acquireHistoryRetentionLease request = send (AcquireHistoryRetentionLease request)++-- | Renew a lease through the mockable 'Store' effect and emit only a committed success.+renewHistoryRetentionLease ::+    (Store :> es) =>+    HistoryRetentionLeaseHandle ->+    HistoryRetentionLeaseDuration ->+    Eff es (Either HistoryRetentionRenewalError HistoryRetentionLease)+renewHistoryRetentionLease handle duration = send (RenewHistoryRetentionLease handle duration)++-- | Release a lease through the mockable 'Store' effect. Repeated release emits no false transition.+releaseHistoryRetentionLease ::+    (Store :> es) =>+    HistoryRetentionLeaseHandle ->+    Eff es HistoryRetentionReleaseResult+releaseHistoryRetentionLease handle = send (ReleaseHistoryRetentionLease handle)++-- | Read the bounded durable inventory through the mockable 'Store' effect.+historyRetentionLeaseInventory ::+    (Store :> es) =>+    HistoryRetentionInventoryQuery ->+    Eff es (Vector HistoryRetentionLease)+historyRetentionLeaseInventory query = send (GetHistoryRetentionLeaseInventory query)++-- | Prune terminal rows through the mockable 'Store' effect.+pruneHistoryRetentionLeases ::+    (Store :> es) =>+    UTCTime ->+    Eff es HistoryRetentionPruneResult+pruneHistoryRetentionLeases cutoff = send (PruneHistoryRetentionLeases cutoff)++{- | Lock one application's stream row in share mode until the surrounding+transaction ends. The returned metadata includes soft-delete and logical+truncate state so a repair can reject incomplete history before mutating its+target. @$all@ is reserved.++When composing this guard with a lease operation, acquire or renew the lease+first and take this guard second.+-}+lockStreamHistoryForReplayTx ::+    StreamName ->+    Tx.Transaction (Either StreamHistoryUnavailable StreamInfo)+lockStreamHistoryForReplayTx stream@(StreamName name)+    | name == "$all" = pure (Left (StreamHistoryReserved stream))+    | otherwise = do+        locked <- Tx.statement name HistoryRetentionSQL.lockStreamHistoryStmt+        pure $ maybe (Left (StreamHistoryNotFound stream)) Right locked++{- | Read a page inside the caller's transaction using the production+exclusive-lower cursor and ascending-order statement. This function deliberately+does not run 'Kiroku.Store.Settings.decodeHook': 'Tx.Transaction' has no 'IO'.+The caller must take 'lockStreamHistoryForReplayTx' earlier in the same+transaction when stable one-stream history is required.+-}+readStreamForwardTx ::+    StreamName ->+    StreamVersion ->+    Int32 ->+    Tx.Transaction (Vector RecordedEvent)+readStreamForwardTx (StreamName name) (StreamVersion cursor) limit =+    Tx.statement (name, cursor, limit) SQL.readStreamForwardStmt

@@ -0,0 +1,91 @@+module Kiroku.Store.HistoryRetention.Internal (+    acquireHistoryRetentionLeaseTx,+    renewHistoryRetentionLeaseTx,+    releaseHistoryRetentionLeaseTx,+    historyRetentionLeaseInventoryTx,+    pruneHistoryRetentionLeasesTx,+    lockHistoryRetentionCoordinatorTx,+    activeHistoryRetentionConflictTx,+    lockAffectedStreamsForHardDeleteTx,+) where++import Data.Int (Int64)+import Data.Time.Clock (UTCTime)+import Data.UUID (UUID)+import Data.Vector (Vector)+import Hasql.Transaction qualified as Tx+import Kiroku.Store.HistoryRetention.SQL qualified as SQL+import Kiroku.Store.HistoryRetention.Types++acquireHistoryRetentionLeaseTx :: HistoryRetentionLeaseRequest -> Tx.Transaction HistoryRetentionLease+acquireHistoryRetentionLeaseTx HistoryRetentionLeaseRequest{owner, reason, duration} =+    Tx.statement+        ( historyRetentionLeaseOwnerText owner+        , historyRetentionLeaseReasonText reason+        , historyRetentionLeaseDurationValue duration+        )+        SQL.acquireLeaseStmt++renewHistoryRetentionLeaseTx ::+    HistoryRetentionLeaseHandle ->+    HistoryRetentionLeaseDuration ->+    Tx.Transaction (Either HistoryRetentionRenewalError HistoryRetentionLease)+renewHistoryRetentionLeaseTx HistoryRetentionLeaseHandle{leaseId, owner = requestedOwner} duration = do+    lockHistoryRetentionCoordinatorTx+    current <- Tx.statement (leaseUuid leaseId) SQL.readLeaseForUpdateStmt+    case current of+        Nothing -> pure (Left HistoryRetentionRenewalUnknown)+        Just lease@HistoryRetentionLease{owner = actualOwner, state}+            | actualOwner /= requestedOwner -> pure (Left HistoryRetentionRenewalOwnerMismatch)+            | state == HistoryRetentionLeaseReleased -> pure (Left HistoryRetentionRenewalReleased)+            | state == HistoryRetentionLeaseExpired -> pure (Left HistoryRetentionRenewalExpired)+            | otherwise ->+                Right+                    <$> Tx.statement+                        (leaseUuid leaseId, historyRetentionLeaseDurationValue duration)+                        SQL.renewLeaseStmt++releaseHistoryRetentionLeaseTx ::+    HistoryRetentionLeaseHandle ->+    Tx.Transaction HistoryRetentionReleaseResult+releaseHistoryRetentionLeaseTx HistoryRetentionLeaseHandle{leaseId, owner = requestedOwner} = do+    lockHistoryRetentionCoordinatorTx+    current <- Tx.statement (leaseUuid leaseId) SQL.readLeaseForUpdateStmt+    case current of+        Nothing -> pure HistoryRetentionReleaseUnknown+        Just lease@HistoryRetentionLease{owner = actualOwner, state}+            | actualOwner /= requestedOwner -> pure HistoryRetentionReleaseOwnerMismatch+            | state == HistoryRetentionLeaseReleased -> pure (HistoryRetentionAlreadyReleased lease)+            | state == HistoryRetentionLeaseExpired -> pure (HistoryRetentionReleaseExpired lease)+            | otherwise ->+                HistoryRetentionReleased+                    <$> Tx.statement (leaseUuid leaseId) SQL.releaseLeaseStmt++historyRetentionLeaseInventoryTx ::+    HistoryRetentionInventoryQuery ->+    Tx.Transaction (Vector HistoryRetentionLease)+historyRetentionLeaseInventoryTx HistoryRetentionInventoryQuery{limit} =+    Tx.statement (historyRetentionInventoryLimitValue limit) SQL.leaseInventoryStmt++pruneHistoryRetentionLeasesTx ::+    UTCTime ->+    Tx.Transaction HistoryRetentionPruneResult+pruneHistoryRetentionLeasesTx cutoff = do+    lockHistoryRetentionCoordinatorTx+    Tx.statement cutoff SQL.pruneLeasesStmt++lockHistoryRetentionCoordinatorTx :: Tx.Transaction ()+lockHistoryRetentionCoordinatorTx = do+    _ <- Tx.statement () SQL.lockCoordinatorStmt+    pure ()++activeHistoryRetentionConflictTx :: Tx.Transaction (Maybe HistoryRetentionConflict)+activeHistoryRetentionConflictTx = Tx.statement () SQL.activeConflictStmt++lockAffectedStreamsForHardDeleteTx :: Int64 -> Tx.Transaction ()+lockAffectedStreamsForHardDeleteTx streamId = do+    _ <- Tx.statement streamId SQL.lockAffectedStreamsForHardDeleteStmt+    pure ()++leaseUuid :: HistoryRetentionLeaseId -> UUID+leaseUuid (HistoryRetentionLeaseId value) = value

@@ -0,0 +1,298 @@+{-# LANGUAGE MultilineStrings #-}++module Kiroku.Store.HistoryRetention.SQL (+    acquireLeaseStmt,+    lockCoordinatorStmt,+    readLeaseForUpdateStmt,+    renewLeaseStmt,+    releaseLeaseStmt,+    leaseInventoryStmt,+    pruneLeasesStmt,+    activeConflictStmt,+    lockStreamHistoryStmt,+    lockAffectedStreamsForHardDeleteStmt,+) where++import Contravariant.Extras (contrazip2, contrazip3)+import Data.Functor.Contravariant ((>$<))+import Data.Int (Int32, Int64)+import Data.Text (Text)+import Data.Time.Clock (DiffTime, UTCTime)+import Data.UUID (UUID)+import Data.Vector (Vector)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Kiroku.Store.HistoryRetention.Types+import Kiroku.Store.Types (GlobalPosition (..), StreamId (..), StreamInfo (..), StreamName (..), StreamVersion (..))++acquireLeaseStmt :: Statement (Text, Text, DiffTime) HistoryRetentionLease+acquireLeaseStmt =+    preparable+        """+        WITH coordinator AS MATERIALIZED (+          SELECT singleton+          FROM history_retention_coordinator+          WHERE singleton+          FOR UPDATE+        ), snapshot AS MATERIALIZED (+          SELECT streams.stream_version AS protected_through,+                 clock_timestamp() AS database_now+          FROM streams+          CROSS JOIN coordinator+          WHERE streams.stream_id = 0+        ), inserted AS (+          INSERT INTO history_retention_leases+            (owner, reason, protected_through, created_at, renewed_at, expires_at)+          SELECT $1, $2, protected_through, database_now, database_now, database_now + $3+          FROM snapshot+          RETURNING lease_id, owner, reason, protected_through,+                    created_at, renewed_at, expires_at, released_at+        )+        SELECT lease_id, owner, reason, protected_through,+               created_at, renewed_at, expires_at, released_at,+               'active'::text+        FROM inserted+        """+        (contrazip3 textParam textParam intervalParam)+        (D.singleRow leaseRow)++lockCoordinatorStmt :: Statement () Bool+lockCoordinatorStmt =+    preparable+        """+        SELECT singleton+        FROM history_retention_coordinator+        WHERE singleton+        FOR UPDATE+        """+        E.noParams+        (D.singleRow (column D.bool))++readLeaseForUpdateStmt :: Statement UUID (Maybe HistoryRetentionLease)+readLeaseForUpdateStmt =+    preparable+        """+        SELECT lease_id, owner, reason, protected_through,+               created_at, renewed_at, expires_at, released_at,+               CASE+                 WHEN released_at IS NOT NULL THEN 'released'+                 WHEN expires_at <= clock_timestamp() THEN 'expired'+                 ELSE 'active'+               END::text+        FROM history_retention_leases+        WHERE lease_id = $1+        FOR UPDATE+        """+        uuidParam+        (D.rowMaybe leaseRow)++renewLeaseStmt :: Statement (UUID, DiffTime) HistoryRetentionLease+renewLeaseStmt =+    preparable+        """+        WITH renewal_time AS MATERIALIZED (+          SELECT clock_timestamp() AS database_now+        ), renewed AS (+          UPDATE history_retention_leases+          SET renewed_at = renewal_time.database_now,+              expires_at = GREATEST(+                history_retention_leases.expires_at,+                renewal_time.database_now + $2+              )+          FROM renewal_time+          WHERE lease_id = $1+          RETURNING lease_id, owner, reason, protected_through,+                    created_at, renewed_at, expires_at, released_at+        )+        SELECT lease_id, owner, reason, protected_through,+               created_at, renewed_at, expires_at, released_at,+               'active'::text+        FROM renewed+        """+        (contrazip2 uuidParam intervalParam)+        (D.singleRow leaseRow)++releaseLeaseStmt :: Statement UUID HistoryRetentionLease+releaseLeaseStmt =+    preparable+        """+        WITH released AS (+          UPDATE history_retention_leases+          SET released_at = clock_timestamp()+          WHERE lease_id = $1+          RETURNING lease_id, owner, reason, protected_through,+                    created_at, renewed_at, expires_at, released_at+        )+        SELECT lease_id, owner, reason, protected_through,+               created_at, renewed_at, expires_at, released_at,+               'released'::text+        FROM released+        """+        uuidParam+        (D.singleRow leaseRow)++leaseInventoryStmt :: Statement Int32 (Vector HistoryRetentionLease)+leaseInventoryStmt =+    preparable+        """+        WITH inventory_time AS MATERIALIZED (+          SELECT clock_timestamp() AS database_now+        )+        SELECT lease_id, owner, reason, protected_through,+               created_at, renewed_at, expires_at, released_at,+               CASE+                 WHEN released_at IS NOT NULL THEN 'released'+                 WHEN expires_at <= inventory_time.database_now THEN 'expired'+                 ELSE 'active'+               END::text+        FROM history_retention_leases+        CROSS JOIN inventory_time+        ORDER BY created_at, lease_id+        LIMIT $1+        """+        int4Param+        (D.rowVector leaseRow)++pruneLeasesStmt :: Statement UTCTime HistoryRetentionPruneResult+pruneLeasesStmt =+    preparable+        """+        WITH released AS (+          DELETE FROM history_retention_leases+          WHERE released_at IS NOT NULL+            AND released_at < $1+          RETURNING 1+        ), expired AS (+          DELETE FROM history_retention_leases+          WHERE released_at IS NULL+            AND expires_at < $1+          RETURNING 1+        )+        SELECT (SELECT count(*) FROM expired),+               (SELECT count(*) FROM released)+        """+        timestamptzParam+        ( D.singleRow+            ( HistoryRetentionPruneResult+                <$> column D.int8+                <*> column D.int8+            )+        )++activeConflictStmt :: Statement () (Maybe HistoryRetentionConflict)+activeConflictStmt =+    preparable+        """+        SELECT count(*), min(expires_at)+        FROM history_retention_leases+        WHERE released_at IS NULL+          AND expires_at > clock_timestamp()+        HAVING count(*) > 0+        """+        E.noParams+        ( D.rowMaybe+            ( HistoryRetentionConflict+                <$> column D.int8+                <*> column D.timestamptz+            )+        )++lockStreamHistoryStmt :: Statement Text (Maybe StreamInfo)+lockStreamHistoryStmt =+    preparable+        """+        SELECT stream_id, stream_name, stream_version,+               created_at, deleted_at, truncate_before+        FROM streams+        WHERE stream_name = $1+        FOR SHARE+        """+        textParam+        (D.rowMaybe streamInfoRow)++lockAffectedStreamsForHardDeleteStmt :: Statement Int64 (Vector Int64)+lockAffectedStreamsForHardDeleteStmt =+    preparable+        """+        WITH affected_streams AS (+          SELECT $1::bigint AS stream_id+          UNION+          SELECT stream_events.stream_id+          FROM stream_events+          WHERE stream_events.original_stream_id = $1+            AND stream_events.stream_id <> 0+        )+        SELECT streams.stream_id+        FROM streams+        JOIN affected_streams USING (stream_id)+        ORDER BY streams.stream_id+        FOR UPDATE OF streams+        """+        int8Param+        (D.rowVector (column D.int8))++leaseRow :: D.Row HistoryRetentionLease+leaseRow =+    makeLease+        <$> column D.uuid+        <*> column D.text+        <*> column D.text+        <*> column D.int8+        <*> column D.timestamptz+        <*> column D.timestamptz+        <*> column D.timestamptz+        <*> D.column (D.nullable D.timestamptz)+        <*> column D.text++makeLease :: UUID -> Text -> Text -> Int64 -> UTCTime -> UTCTime -> UTCTime -> Maybe UTCTime -> Text -> HistoryRetentionLease+makeLease leaseUuid ownerText reasonText frontier created renewed expires released stateText =+    HistoryRetentionLease+        { leaseId = HistoryRetentionLeaseId leaseUuid+        , owner = requireValidated (mkHistoryRetentionLeaseOwner ownerText)+        , reason = requireValidated (mkHistoryRetentionLeaseReason reasonText)+        , protectedThrough = GlobalPosition frontier+        , createdAt = created+        , renewedAt = renewed+        , expiresAt = expires+        , releasedAt = released+        , state = case stateText of+            "active" -> HistoryRetentionLeaseActive+            "expired" -> HistoryRetentionLeaseExpired+            "released" -> HistoryRetentionLeaseReleased+            unexpected -> error ("unknown history retention lease state from database: " <> show unexpected)+        }++streamInfoRow :: D.Row StreamInfo+streamInfoRow =+    StreamInfo+        <$> (StreamId <$> column D.int8)+        <*> (StreamName <$> column D.text)+        <*> (StreamVersion <$> column D.int8)+        <*> column D.timestamptz+        <*> D.column (D.nullable D.timestamptz)+        <*> (StreamVersion <$> column D.int8)++requireValidated :: Either error value -> value+requireValidated = either (const (error "database returned an invalid history retention lease")) (\value -> value)++column :: D.Value value -> D.Row value+column = D.column . D.nonNullable++textParam :: E.Params Text+textParam = E.param (E.nonNullable E.text)++uuidParam :: E.Params UUID+uuidParam = E.param (E.nonNullable E.uuid)++int4Param :: E.Params Int32+int4Param = E.param (E.nonNullable E.int4)++int8Param :: E.Params Int64+int8Param = E.param (E.nonNullable E.int8)++intervalParam :: E.Params DiffTime+intervalParam = E.param (E.nonNullable E.interval)++timestamptzParam :: E.Params UTCTime+timestamptzParam = E.param (E.nonNullable E.timestamptz)

@@ -0,0 +1,180 @@+module Kiroku.Store.HistoryRetention.Types (+    HistoryRetentionLeaseId (..),+    HistoryRetentionLeaseOwner,+    HistoryRetentionLeaseReason,+    HistoryRetentionLeaseDuration,+    HistoryRetentionLeaseRequest (..),+    HistoryRetentionLeaseHandle (..),+    HistoryRetentionLeaseState (..),+    HistoryRetentionLease (..),+    HistoryRetentionRenewalError (..),+    HistoryRetentionReleaseResult (..),+    HistoryRetentionConflict (..),+    HistoryRetentionInventoryLimit,+    HistoryRetentionInventoryQuery (..),+    HistoryRetentionPruneResult (..),+    HistoryRetentionRequestError (..),+    HistoryRetentionInventoryError (..),+    StreamHistoryUnavailable (..),+    mkHistoryRetentionLeaseOwner,+    mkHistoryRetentionLeaseReason,+    mkHistoryRetentionLeaseDuration,+    mkHistoryRetentionInventoryLimit,+    historyRetentionLeaseOwnerText,+    historyRetentionLeaseReasonText,+    historyRetentionLeaseDurationValue,+    historyRetentionInventoryLimitValue,+    maxHistoryRetentionLeaseDuration,+) where++import Data.ByteString qualified as ByteString+import Data.Int (Int32, Int64)+import Data.Text (Text)+import Data.Text.Encoding qualified as Text+import Data.Time.Clock (DiffTime, UTCTime, secondsToDiffTime)+import Data.UUID (UUID)+import GHC.Generics (Generic)+import Kiroku.Store.Types (GlobalPosition, StreamName)++-- | Database-generated identity of one durable retention lease.+newtype HistoryRetentionLeaseId = HistoryRetentionLeaseId UUID+    deriving stock (Eq, Ord, Show, Generic)++-- | Validated operational owner label (1 through 512 UTF-8 bytes).+newtype HistoryRetentionLeaseOwner = HistoryRetentionLeaseOwner Text+    deriving stock (Eq, Ord, Show, Generic)++-- | Validated human-readable reason (1 through 2,048 UTF-8 bytes).+newtype HistoryRetentionLeaseReason = HistoryRetentionLeaseReason Text+    deriving stock (Eq, Ord, Show, Generic)++-- | Requested remaining lifetime, from one second through one hour.+newtype HistoryRetentionLeaseDuration = HistoryRetentionLeaseDuration DiffTime+    deriving stock (Eq, Ord, Show, Generic)++data HistoryRetentionLeaseRequest = HistoryRetentionLeaseRequest+    { owner :: !HistoryRetentionLeaseOwner+    , reason :: !HistoryRetentionLeaseReason+    , duration :: !HistoryRetentionLeaseDuration+    }+    deriving stock (Eq, Show, Generic)++data HistoryRetentionLeaseHandle = HistoryRetentionLeaseHandle+    { leaseId :: !HistoryRetentionLeaseId+    , owner :: !HistoryRetentionLeaseOwner+    }+    deriving stock (Eq, Show, Generic)++data HistoryRetentionLeaseState+    = HistoryRetentionLeaseActive+    | HistoryRetentionLeaseExpired+    | HistoryRetentionLeaseReleased+    deriving stock (Eq, Ord, Show, Generic)++data HistoryRetentionLease = HistoryRetentionLease+    { leaseId :: !HistoryRetentionLeaseId+    , owner :: !HistoryRetentionLeaseOwner+    , reason :: !HistoryRetentionLeaseReason+    , protectedThrough :: !GlobalPosition+    , createdAt :: !UTCTime+    , renewedAt :: !UTCTime+    , expiresAt :: !UTCTime+    , releasedAt :: !(Maybe UTCTime)+    , state :: !HistoryRetentionLeaseState+    }+    deriving stock (Eq, Show, Generic)++data HistoryRetentionRenewalError+    = HistoryRetentionRenewalUnknown+    | HistoryRetentionRenewalOwnerMismatch+    | HistoryRetentionRenewalExpired+    | HistoryRetentionRenewalReleased+    deriving stock (Eq, Show, Generic)++data HistoryRetentionReleaseResult+    = HistoryRetentionReleased !HistoryRetentionLease+    | HistoryRetentionAlreadyReleased !HistoryRetentionLease+    | HistoryRetentionReleaseExpired !HistoryRetentionLease+    | HistoryRetentionReleaseUnknown+    | HistoryRetentionReleaseOwnerMismatch+    deriving stock (Eq, Show, Generic)++data HistoryRetentionConflict = HistoryRetentionConflict+    { activeLeaseCount :: !Int64+    , earliestExpiry :: !UTCTime+    }+    deriving stock (Eq, Show, Generic)++-- | Validated maximum number of inventory rows (1 through 1,000).+newtype HistoryRetentionInventoryLimit = HistoryRetentionInventoryLimit Int32+    deriving stock (Eq, Ord, Show, Generic)++data HistoryRetentionInventoryQuery = HistoryRetentionInventoryQuery+    { limit :: !HistoryRetentionInventoryLimit+    }+    deriving stock (Eq, Show, Generic)++data HistoryRetentionPruneResult = HistoryRetentionPruneResult+    { expiredPruned :: !Int64+    , releasedPruned :: !Int64+    }+    deriving stock (Eq, Show, Generic)++data HistoryRetentionRequestError+    = HistoryRetentionLeaseOwnerEmpty+    | HistoryRetentionLeaseOwnerTooLong !Int+    | HistoryRetentionLeaseReasonEmpty+    | HistoryRetentionLeaseReasonTooLong !Int+    | HistoryRetentionLeaseDurationOutOfRange !DiffTime+    deriving stock (Eq, Show, Generic)++data HistoryRetentionInventoryError+    = HistoryRetentionInventoryLimitOutOfRange !Int32+    deriving stock (Eq, Show, Generic)++data StreamHistoryUnavailable+    = StreamHistoryNotFound !StreamName+    | StreamHistoryReserved !StreamName+    deriving stock (Eq, Show, Generic)++mkHistoryRetentionLeaseOwner :: Text -> Either HistoryRetentionRequestError HistoryRetentionLeaseOwner+mkHistoryRetentionLeaseOwner value+    | bytes == 0 = Left HistoryRetentionLeaseOwnerEmpty+    | bytes > 512 = Left (HistoryRetentionLeaseOwnerTooLong bytes)+    | otherwise = Right (HistoryRetentionLeaseOwner value)+  where+    bytes = ByteString.length (Text.encodeUtf8 value)++mkHistoryRetentionLeaseReason :: Text -> Either HistoryRetentionRequestError HistoryRetentionLeaseReason+mkHistoryRetentionLeaseReason value+    | bytes == 0 = Left HistoryRetentionLeaseReasonEmpty+    | bytes > 2048 = Left (HistoryRetentionLeaseReasonTooLong bytes)+    | otherwise = Right (HistoryRetentionLeaseReason value)+  where+    bytes = ByteString.length (Text.encodeUtf8 value)++mkHistoryRetentionLeaseDuration :: DiffTime -> Either HistoryRetentionRequestError HistoryRetentionLeaseDuration+mkHistoryRetentionLeaseDuration value+    | value < secondsToDiffTime 1 = Left (HistoryRetentionLeaseDurationOutOfRange value)+    | value > maxHistoryRetentionLeaseDuration = Left (HistoryRetentionLeaseDurationOutOfRange value)+    | otherwise = Right (HistoryRetentionLeaseDuration value)++mkHistoryRetentionInventoryLimit :: Int32 -> Either HistoryRetentionInventoryError HistoryRetentionInventoryLimit+mkHistoryRetentionInventoryLimit value+    | value < 1 || value > 1000 = Left (HistoryRetentionInventoryLimitOutOfRange value)+    | otherwise = Right (HistoryRetentionInventoryLimit value)++historyRetentionLeaseOwnerText :: HistoryRetentionLeaseOwner -> Text+historyRetentionLeaseOwnerText (HistoryRetentionLeaseOwner value) = value++historyRetentionLeaseReasonText :: HistoryRetentionLeaseReason -> Text+historyRetentionLeaseReasonText (HistoryRetentionLeaseReason value) = value++historyRetentionLeaseDurationValue :: HistoryRetentionLeaseDuration -> DiffTime+historyRetentionLeaseDurationValue (HistoryRetentionLeaseDuration value) = value++historyRetentionInventoryLimitValue :: HistoryRetentionInventoryLimit -> Int32+historyRetentionInventoryLimitValue (HistoryRetentionInventoryLimit value) = value++maxHistoryRetentionLeaseDuration :: DiffTime+maxHistoryRetentionLeaseDuration = secondsToDiffTime 3600

@@ -73,6 +73,13 @@   authorization layer before they reach this function. Reading the   @protect_deletion@ trigger as a security boundary is incorrect. +Replay-history retention is an independent safety layer. While any+'Kiroku.Store.HistoryRetention.Types.HistoryRetentionLease' is active,+supported hard delete returns+'Kiroku.Store.Error.HistoryRetentionActive' before changing rows. A+GUC-enabled direct @DELETE@ or @TRUNCATE@ receives SQLSTATE @KR001@. There is+no ordinary bypass: release the lease or wait for its database-derived expiry.+ == Event preservation semantics  The interpreter cleans up junction rows ('stream_events') first,@@ -80,6 +87,12 @@ streams from this one's hard-deleted source junctions are removed; events linked to streams /not/ owned by this deletion are preserved. +Before deletion, Kiroku locks the target and every stream containing a link to+one of the target's originated events in ascending internal ID order. This+serializes hard delete with+'Kiroku.Store.HistoryRetention.lockStreamHistoryForReplayTx' guards, including+a guard held on a linked stream.+ == Result  Returns @Just streamId@ on success, @Nothing@ if the stream did not@@ -87,8 +100,8 @@ 'Kiroku.Store.Error.ReservedStreamName'. There is no \"undo\" — for reversible deletes use 'softDeleteStream' instead. The deletion emits no in-band audit row; operators relying on an audit log must capture hard-deletes through-the connection-pool observation handler (see-'Kiroku.Store.Connection.ConnectionSettings.observationHandler') or+the store event handler (see+'Kiroku.Store.Connection.ConnectionSettingsM.eventHandler') or record an application-level event /before/ calling this function. -} hardDeleteStream ::

@@ -20,7 +20,14 @@ * Hard-delete issuance (a fail-safe audit signal — see   @docs\/PRODUCTION-DEPLOYMENT.md@ for the recommended in-band audit   pattern).+* Committed replay-history lease acquisition, renewal, actual release, pruning,+  and supported hard-delete refusal. +Lease reasons remain in the durable inventory and are deliberately absent from+these process-local events. Do not turn free-form reasons into metrics labels.+Direct @Tx.Transaction@ combinators cannot emit process-local events; the+mockable effect wrappers emit only after their database transaction finishes.+ Wire 'Kiroku.Store.Connection.ConnectionSettingsM.eventHandler' to a callback that forwards to your structured logger or metrics pipeline. The callback runs synchronously on the emit-site thread (notifier loop,@@ -47,7 +54,9 @@ import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO) import Data.Foldable (for_) import Data.Int (Int32)+import Data.Time.Clock (UTCTime) import Hasql.Pool (UsageError)+import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict, HistoryRetentionLeaseId, HistoryRetentionLeaseOwner, HistoryRetentionPruneResult) import Kiroku.Store.Subscription.Fsm (DeadLetterReason (..), SubscriptionStopReason (..)) import Kiroku.Store.Subscription.Types (     CheckpointInitialization,@@ -213,6 +222,18 @@       stream did not exist.       -}       KirokuEventHardDeleteIssued !StreamName !StreamId+    | {- | A lease acquisition committed. The lease row is the durable audit+      evidence; this process-local event contains no event payload.+      -}+      KirokuEventHistoryRetentionLeaseAcquired !HistoryRetentionLeaseId !HistoryRetentionLeaseOwner !GlobalPosition !UTCTime+    | -- | A live lease renewal committed; reason text is not copied into the event.+      KirokuEventHistoryRetentionLeaseRenewed !HistoryRetentionLeaseId !HistoryRetentionLeaseOwner !UTCTime+    | -- | A previously active lease was actually released and committed.+      KirokuEventHistoryRetentionLeaseReleased !HistoryRetentionLeaseId !HistoryRetentionLeaseOwner !UTCTime+    | -- | A prune operation committed with the reported terminal-row counts.+      KirokuEventHistoryRetentionLeasesPruned !HistoryRetentionPruneResult+    | -- | Supported hard delete was refused before changing history.+      KirokuEventHardDeleteHistoryRetentionConflict !StreamName !HistoryRetentionConflict     deriving stock (Show)  {- | Invoke the optional observability handler, dropping any synchronous exception

@@ -20,6 +20,11 @@   transaction. This is the primary API for @keiro@-style projection   consumers. +Replay-history leases and one-stream repair guards are exposed separately by+'Kiroku.Store.HistoryRetention' as @Tx.Transaction@ combinators. When composing+both in one body, acquire or renew the lease before taking the stream guard.+The transactional read deliberately bypasses the IO-only decode hook.+ The @-NoRetry@ variants use 'Hasql.Transaction.Sessions.transactionNoRetry' under the hood; the default variants use 'Hasql.Transaction.Sessions.transaction', which

@@ -38,6 +38,8 @@ import Test.EventTypeFilter qualified as EventTypeFilter import Test.FailureInjection qualified as FailureInjection import Test.Helpers+import Test.HistoryRetention qualified as HistoryRetention+import Test.HistoryRetentionMock qualified as HistoryRetentionMock import Test.Hspec import Test.InterpreterHooks qualified as InterpreterHooks import Test.NotifyGuard qualified as NotifyGuard@@ -49,6 +51,7 @@ import Test.ReadStream qualified as ReadStream import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing import Test.StreamBridgeTermination qualified as StreamBridgeTermination+import Test.StreamHistoryGuard qualified as StreamHistoryGuard import Test.StreamNameLookup qualified as StreamNameLookup import Test.SubscriptionCheckpointInitialization qualified as SubscriptionCheckpointInitialization import Test.SubscriptionCheckpointInitializationMock qualified as SubscriptionCheckpointInitializationMock@@ -72,11 +75,14 @@     Properties.spec     Concurrency.spec     FailureInjection.spec+    HistoryRetention.spec+    HistoryRetentionMock.spec     Transaction.spec     TruncateBefore.spec     ReadStream.spec     StreamNameLookup.spec     StreamBridgeTermination.spec+    StreamHistoryGuard.spec     InterpreterHooks.spec     Causation.spec     ConsumerGroupSql.spec

@@ -0,0 +1,376 @@+{-# LANGUAGE MultilineStrings #-}+{-# LANGUAGE NumericUnderscores #-}++module Test.HistoryRetention (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Lens ((&), (.~), (^.))+import Data.Aeson qualified as Aeson+import Data.Either (isLeft)+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.Int (Int32, Int64)+import Data.Text qualified as Text+import Data.Time.Clock (addUTCTime, diffUTCTime, getCurrentTime, secondsToDiffTime)+import Data.Vector qualified as Vector+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Errors qualified as Errors+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable, unpreparable)+import Hasql.Transaction qualified as Tx+import Hasql.Transaction.Sessions qualified as TxSessions+import Kiroku.Store+import Test.Helpers (countEvents, makeEvent, withTestStore, withTestStoreSettings)+import Test.Hspec++spec :: Spec+spec = describe "history retention" $ do+    it "validates UTF-8 byte lengths, duration bounds, and inventory bounds" $ do+        mkHistoryRetentionLeaseOwner "" `shouldBe` Left HistoryRetentionLeaseOwnerEmpty+        mkHistoryRetentionLeaseOwner (Text.replicate 256 "é") `shouldSatisfy` either (const False) (const True)+        mkHistoryRetentionLeaseOwner (Text.replicate 257 "é")+            `shouldBe` Left (HistoryRetentionLeaseOwnerTooLong 514)+        mkHistoryRetentionLeaseReason "" `shouldBe` Left HistoryRetentionLeaseReasonEmpty+        mkHistoryRetentionLeaseReason (Text.replicate 2049 "x")+            `shouldBe` Left (HistoryRetentionLeaseReasonTooLong 2049)+        mkHistoryRetentionLeaseDuration (secondsToDiffTime 0) `shouldSatisfy` isLeft+        mkHistoryRetentionLeaseDuration (secondsToDiffTime 1) `shouldSatisfy` either (const False) (const True)+        mkHistoryRetentionLeaseDuration (secondsToDiffTime 3600) `shouldSatisfy` either (const False) (const True)+        mkHistoryRetentionLeaseDuration (secondsToDiffTime 3601) `shouldSatisfy` isLeft+        mkHistoryRetentionInventoryLimit 0 `shouldBe` Left (HistoryRetentionInventoryLimitOutOfRange 0)+        mkHistoryRetentionInventoryLimit 1000 `shouldSatisfy` either (const False) (const True)+        mkHistoryRetentionInventoryLimit 1001 `shouldBe` Left (HistoryRetentionInventoryLimitOutOfRange 1001)++    it "captures the authoritative frontier and database-derived expiry" $+        withTestStore $ \store -> do+            Right appended <-+                runStoreIO store $+                    appendToStream+                        (StreamName "history-retention-frontier")+                        NoStream+                        [ makeEvent "One" (Aeson.object [])+                        , makeEvent "Two" (Aeson.object [])+                        , makeEvent "Three" (Aeson.object [])+                        ]+            lease <- runTx store (acquireHistoryRetentionLeaseTx (request "rebuild" "frontier" 60))+            lease ^. #protectedThrough `shouldBe` appended ^. #globalPosition+            realToFrac (diffUTCTime (lease ^. #expiresAt) (lease ^. #createdAt))+                `shouldBe` secondsToDiffTime 60+            lease ^. #state `shouldBe` HistoryRetentionLeaseActive+            rows <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 10))+            rows `shouldBe` Vector.singleton lease++    it "rolls back acquisition when the surrounding transaction is condemned" $+        withTestStore $ \store -> do+            _ <- runTx store $ do+                lease <- acquireHistoryRetentionLeaseTx (request "rollback" "condemned" 60)+                Tx.condemn+                pure lease+            rows <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 10))+            rows `shouldBe` Vector.empty++    it "renews only the matching active owner and never shortens expiry" $+        withTestStore $ \store -> do+            lease <- runTx store (acquireHistoryRetentionLeaseTx (request "owner-a" "renew" 30))+            let handle = leaseHandle lease+                wrongHandle = HistoryRetentionLeaseHandle (lease ^. #leaseId) (validatedOwner "owner-b")+            mismatch <- runTx store (renewHistoryRetentionLeaseTx wrongHandle (validatedDuration 60))+            mismatch `shouldBe` Left HistoryRetentionRenewalOwnerMismatch+            renewed <- runTx store (renewHistoryRetentionLeaseTx handle (validatedDuration 60))+            case renewed of+                Left err -> expectationFailure ("renewal failed: " <> show err)+                Right value -> do+                    value ^. #expiresAt `shouldSatisfy` (> lease ^. #expiresAt)+                    value ^. #renewedAt `shouldSatisfy` (>= lease ^. #renewedAt)++    it "derives expiry without a worker and refuses resurrection" $+        withTestStore $ \store -> do+            lease <- runTx store (acquireHistoryRetentionLeaseTx (request "crashed" "expiry" 1))+            threadDelay 1_100_000+            rows <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 10))+            fmap (^. #state) (Vector.toList rows) `shouldBe` [HistoryRetentionLeaseExpired]+            renewed <- runTx store (renewHistoryRetentionLeaseTx (leaseHandle lease) (validatedDuration 60))+            renewed `shouldBe` Left HistoryRetentionRenewalExpired+            released <- runTx store (releaseHistoryRetentionLeaseTx (leaseHandle lease))+            case released of+                HistoryRetentionReleaseExpired expiredLease ->+                    expiredLease ^. #state `shouldBe` HistoryRetentionLeaseExpired+                other -> expectationFailure ("expected expired release result, got " <> show other)++    it "releases idempotently and leaves another simultaneous lease active" $+        withTestStore $ \store -> do+            first <- runTx store (acquireHistoryRetentionLeaseTx (request "first" "release" 60))+            second <- runTx store (acquireHistoryRetentionLeaseTx (request "second" "release" 60))+            released <- runTx store (releaseHistoryRetentionLeaseTx (leaseHandle first))+            released `shouldSatisfy` \case HistoryRetentionReleased{} -> True; _ -> False+            repeated <- runTx store (releaseHistoryRetentionLeaseTx (leaseHandle first))+            repeated `shouldSatisfy` \case HistoryRetentionAlreadyReleased{} -> True; _ -> False+            rows <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 10))+            fmap (^. #state) (Vector.toList rows)+                `shouldBe` [HistoryRetentionLeaseReleased, HistoryRetentionLeaseActive]+            Vector.last rows ^. #leaseId `shouldBe` second ^. #leaseId++    it "bounds inventory and prunes only terminal rows older than the cutoff" $+        withTestStore $ \store -> do+            first <- runTx store (acquireHistoryRetentionLeaseTx (request "first" "prune" 60))+            _ <- runTx store (acquireHistoryRetentionLeaseTx (request "second" "keep" 60))+            _ <- runTx store (releaseHistoryRetentionLeaseTx (leaseHandle first))+            bounded <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 1))+            Vector.length bounded `shouldBe` 1+            cutoff <- addUTCTime 1 <$> getCurrentTime+            pruned <- runTx store (pruneHistoryRetentionLeasesTx cutoff)+            pruned `shouldBe` HistoryRetentionPruneResult 0 1+            remaining <- runTx store (historyRetentionLeaseInventoryTx (inventoryQuery 10))+            fmap (^. #state) (Vector.toList remaining) `shouldBe` [HistoryRetentionLeaseActive]++    it "emits committed effect transitions once and no false repeated-release event" $ do+        observed <- newIORef ([] :: [KirokuEvent])+        withTestStoreSettings+            (& #eventHandler .~ Just (\event -> modifyIORef' observed (event :)))+            $ \store -> do+                Right lease <- runStoreIO store $ acquireHistoryRetentionLease (request "events" "not-a-label" 60)+                Right (Right _) <-+                    runStoreIO store $+                        renewHistoryRetentionLease (leaseHandle lease) (validatedDuration 120)+                Right HistoryRetentionReleased{} <-+                    runStoreIO store $+                        releaseHistoryRetentionLease (leaseHandle lease)+                Right HistoryRetentionAlreadyReleased{} <-+                    runStoreIO store $+                        releaseHistoryRetentionLease (leaseHandle lease)+                cutoff <- addUTCTime 1 <$> getCurrentTime+                Right (HistoryRetentionPruneResult 0 1) <-+                    runStoreIO store $+                        pruneHistoryRetentionLeases cutoff+                pure ()+        events <- readIORef observed+        length [() | KirokuEventHistoryRetentionLeaseAcquired{} <- events] `shouldBe` 1+        length [() | KirokuEventHistoryRetentionLeaseRenewed{} <- events] `shouldBe` 1+        length [() | KirokuEventHistoryRetentionLeaseReleased{} <- events] `shouldBe` 1+        length [() | KirokuEventHistoryRetentionLeasesPruned{} <- events] `shouldBe` 1++    it "returns and emits a typed hard-delete conflict without changing history while any lease is active" $ do+        observed <- newIORef ([] :: [KirokuEvent])+        withTestStoreSettings+            (& #eventHandler .~ Just (\event -> modifyIORef' observed (event :)))+            $ \store -> do+                let stream = StreamName "history-retention-hard-delete"+                Right _ <- runStoreIO store $ appendToStream stream NoStream [makeEvent "Protected" (Aeson.object [])]+                before <- countEvents store+                Right first <- runStoreIO store $ acquireHistoryRetentionLease (request "first" "protect" 60)+                Right second <- runStoreIO store $ acquireHistoryRetentionLease (request "second" "protect" 60)+                blocked <- runStoreIO store $ hardDeleteStream stream+                case blocked of+                    Left (HistoryRetentionActive actual HistoryRetentionConflict{activeLeaseCount}) -> do+                        actual `shouldBe` stream+                        activeLeaseCount `shouldBe` 2+                    other -> expectationFailure ("expected typed retention conflict, got " <> show other)+                countEvents store `shouldReturn` before+                Right (Just _) <- runStoreIO store $ getStream stream+                Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle first)+                stillBlocked <- runStoreIO store $ hardDeleteStream stream+                stillBlocked `shouldSatisfy` \case Left HistoryRetentionActive{} -> True; _ -> False+                Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle second)+                deleted <- runStoreIO store (hardDeleteStream stream)+                deleted `shouldSatisfy` \case Right (Just _) -> True; _ -> False+        events <- readIORef observed+        let conflictCounts =+                [ activeLeaseCount+                | KirokuEventHardDeleteHistoryRetentionConflict _ HistoryRetentionConflict{activeLeaseCount} <- reverse events+                ]+        conflictCounts `shouldBe` [2, 1]++    describe "history retention raw SQL" $ do+        it "serializes lease-first acquisition ahead of raw deletion" $+            withTestStore $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "raw-race-lease-first") NoStream [makeEvent "Raw" (Aeson.object [])]+                before <- countStreamEvents store+                acquisition <- Async.async $ runStoreIO store $ runTransaction $ do+                    lease <- acquireHistoryRetentionLeaseTx (request "raw-race" "lease-first" 60)+                    _ <- Tx.statement () holdCoordinatorStmt+                    pure lease+                waitForCoordinatorPhase store 100+                deletion <- Async.async (runRawDestruction store rawDeleteStreamEventsStmt)+                threadDelay 50_000+                Async.poll deletion >>= \case+                    Nothing -> pure ()+                    Just _ -> expectationFailure "raw deletion completed while lease acquisition held the coordinator"+                acquired <- waitWithin "lease-first acquisition" acquisition+                acquired `shouldSatisfy` \case Right HistoryRetentionLease{} -> True; _ -> False+                rejected <- waitWithin "lease-first raw deletion" deletion+                rejected `shouldSatisfy` hasSqlState "KR001"+                countStreamEvents store `shouldReturn` before++        it "serializes delete-first maintenance ahead of post-delete acquisition" $+            withTestStore $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "raw-race-delete-first") NoStream [makeEvent "Raw" (Aeson.object [])]+                deletion <- Async.async (runRawDestructionHeld store rawDeleteStreamEventsStmt)+                waitForCoordinatorPhase store 100+                acquisition <- Async.async $ runStoreIO store $ acquireHistoryRetentionLease (request "raw-race" "delete-first" 60)+                threadDelay 50_000+                Async.poll acquisition >>= \case+                    Nothing -> pure ()+                    Just _ -> expectationFailure "lease acquisition completed while raw deletion held the coordinator"+                waitWithin "delete-first raw deletion" deletion `shouldReturn` Right ()+                acquired <- waitWithin "delete-first acquisition" acquisition+                acquired `shouldSatisfy` \case Right HistoryRetentionLease{} -> True; _ -> False+                countStreamEvents store `shouldReturn` 0++        it "rejects GUC-enabled DELETE with KR001 and permits it after release" $+            withTestStore $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "raw-delete") NoStream (replicate 2 (makeEvent "Raw" (Aeson.object [])))+                before <- countStreamEvents store+                Right lease <- runStoreIO store $ acquireHistoryRetentionLease (request "raw" "delete" 60)+                rejected <- runRawDestruction store rawDeleteStreamEventsStmt+                rejected `shouldSatisfy` hasSqlState "KR001"+                countStreamEvents store `shouldReturn` before+                Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle lease)+                runRawDestruction store rawDeleteStreamEventsStmt `shouldReturn` Right ()+                countStreamEvents store `shouldReturn` 0++        it "rejects GUC-enabled TRUNCATE with KR001 and permits it after release" $+            withTestStore $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "raw-truncate") NoStream [makeEvent "Raw" (Aeson.object [])]+                before <- countEvents store+                Right lease <- runStoreIO store $ acquireHistoryRetentionLease (request "raw" "truncate" 60)+                rejected <- runRawDestruction store rawTruncateDataStmt+                rejected `shouldSatisfy` hasSqlState "KR001"+                countEvents store `shouldReturn` before+                Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle lease)+                runRawDestruction store rawTruncateDataStmt `shouldReturn` Right ()+                countEvents store `shouldReturn` 0++        it "permits GUC-enabled maintenance after passive expiry" $+            withTestStore $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "raw-expiry") NoStream [makeEvent "Raw" (Aeson.object [])]+                Right _ <- runStoreIO store $ acquireHistoryRetentionLease (request "raw" "expiry" 1)+                threadDelay 1_100_000+                runRawDestruction store rawDeleteStreamEventsStmt `shouldReturn` Right ()++runTx :: KirokuStore -> Tx.Transaction value -> IO value+runTx store transaction = do+    result <- runStoreIO store (runTransaction transaction)+    case result of+        Left err -> expectationFailure ("history retention transaction failed: " <> show err) >> error "unreachable"+        Right value -> pure value++request :: Text.Text -> Text.Text -> Integer -> HistoryRetentionLeaseRequest+request ownerText reasonText seconds =+    HistoryRetentionLeaseRequest+        { owner = validatedOwner ownerText+        , reason = either (error . show) (\value -> value) (mkHistoryRetentionLeaseReason reasonText)+        , duration = validatedDuration seconds+        }++validatedOwner :: Text.Text -> HistoryRetentionLeaseOwner+validatedOwner = either (error . show) (\value -> value) . mkHistoryRetentionLeaseOwner++validatedDuration :: Integer -> HistoryRetentionLeaseDuration+validatedDuration = either (error . show) (\value -> value) . mkHistoryRetentionLeaseDuration . secondsToDiffTime++inventoryQuery :: Int32 -> HistoryRetentionInventoryQuery+inventoryQuery = HistoryRetentionInventoryQuery . either (error . show) (\value -> value) . mkHistoryRetentionInventoryLimit++leaseHandle :: HistoryRetentionLease -> HistoryRetentionLeaseHandle+leaseHandle lease = HistoryRetentionLeaseHandle (lease ^. #leaseId) (lease ^. #owner)++runRawDestruction :: KirokuStore -> Statement () () -> IO (Either Pool.UsageError ())+runRawDestruction store statement =+    Pool.use (store ^. #pool) $+        TxSessions.transaction TxSessions.ReadCommitted TxSessions.Write $ do+            Tx.sql "SET LOCAL kiroku.enable_hard_deletes = 'on'"+            Tx.statement () statement++runRawDestructionHeld :: KirokuStore -> Statement () () -> IO (Either Pool.UsageError ())+runRawDestructionHeld store statement =+    Pool.use (store ^. #pool) $+        TxSessions.transaction TxSessions.ReadCommitted TxSessions.Write $ do+            Tx.sql "SET LOCAL kiroku.enable_hard_deletes = 'on'"+            Tx.statement () statement+            _ <- Tx.statement () holdCoordinatorStmt+            pure ()++waitForCoordinatorPhase :: KirokuStore -> Int -> IO ()+waitForCoordinatorPhase _ 0 = expectationFailure "coordinator holder never reached its held phase"+waitForCoordinatorPhase store attempts = do+    result <- Pool.use (store ^. #pool) (Session.statement () coordinatorActiveStmt)+    case result of+        Right True -> pure ()+        Right False -> threadDelay 10_000 >> waitForCoordinatorPhase store (attempts - 1)+        Left err -> expectationFailure ("could not observe coordinator phase: " <> show err)++waitWithin :: String -> Async.Async value -> IO value+waitWithin label action = do+    result <- Async.race (threadDelay 2_000_000) (Async.wait action)+    case result of+        Left () -> do+            Async.cancel action+            expectationFailure (label <> " timed out")+            error "unreachable"+        Right value -> pure value++hasSqlState :: Text.Text -> Either Pool.UsageError value -> Bool+hasSqlState expected = \case+    Left+        ( Pool.SessionUsageError+                ( Errors.StatementSessionError+                        _+                        _+                        _+                        _+                        _+                        (Errors.ServerStatementError (Errors.ServerError actual _ _ _ _))+                    )+            ) -> actual == expected+    _ -> False++countStreamEvents :: KirokuStore -> IO Int64+countStreamEvents store = do+    result <- Pool.use (store ^. #pool) (TxSessions.transaction TxSessions.ReadCommitted TxSessions.Read (Tx.statement () countStreamEventsStmt))+    case result of+        Left err -> expectationFailure ("could not count stream junctions: " <> show err) >> error "unreachable"+        Right value -> pure value++countStreamEventsStmt :: Statement () Int64+countStreamEventsStmt =+    unpreparable+        "SELECT count(*) FROM stream_events"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.int8)))++rawDeleteStreamEventsStmt :: Statement () ()+rawDeleteStreamEventsStmt =+    unpreparable+        "DELETE FROM stream_events"+        E.noParams+        D.noResult++rawTruncateDataStmt :: Statement () ()+rawTruncateDataStmt =+    unpreparable+        "TRUNCATE dead_letters, stream_events, events"+        E.noParams+        D.noResult++holdCoordinatorStmt :: Statement () Bool+holdCoordinatorStmt =+    preparable+        "SELECT pg_sleep(0.4) IS NULL /* history-retention-coordinator-race */"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))++coordinatorActiveStmt :: Statement () Bool+coordinatorActiveStmt =+    preparable+        """+        SELECT EXISTS (+          SELECT 1+          FROM pg_stat_activity+          WHERE state = 'active'+            AND query = 'SELECT pg_sleep(0.4) IS NULL /* history-retention-coordinator-race */'+        )+        """+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))

@@ -0,0 +1,106 @@+module Test.HistoryRetentionMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Text (Text)+import Data.Time.Calendar (fromGregorian)+import Data.Time.Clock (UTCTime (..), secondsToDiffTime)+import Data.UUID qualified as UUID+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store.Effect (Store (..))+import Kiroku.Store.HistoryRetention+import Kiroku.Store.Types (GlobalPosition (..))+import Test.Hspec++spec :: Spec+spec = describe "history retention mock" $ do+    it "dispatches every closed lease operation exactly once" $ do+        calls <- newIORef ([] :: [Text])+        actual <- runEff $ runMock calls $ do+            acquired <- acquireHistoryRetentionLease sampleRequest+            renewed <- renewHistoryRetentionLease sampleHandle sampleDuration+            released <- releaseHistoryRetentionLease sampleHandle+            inventory <- historyRetentionLeaseInventory sampleQuery+            pruned <- pruneHistoryRetentionLeases sampleTime+            pure (acquired, renewed, released, inventory, pruned)+        actual+            `shouldBe` ( sampleLease+                       , Right sampleLease+                       , HistoryRetentionReleased sampleLease+                       , Vector.singleton sampleLease+                       , HistoryRetentionPruneResult 2 3+                       )+        readIORef calls+            `shouldReturn` ["acquire", "renew", "release", "inventory", "prune"]++runMock :: (IOE :> es) => IORef [Text] -> Eff (Store : es) a -> Eff es a+runMock calls = interpret_ $ \case+    AcquireHistoryRetentionLease request -> do+        liftIO $ request `shouldBe` sampleRequest+        record calls "acquire"+        pure sampleLease+    RenewHistoryRetentionLease handle duration -> do+        liftIO $ (handle, duration) `shouldBe` (sampleHandle, sampleDuration)+        record calls "renew"+        pure (Right sampleLease)+    ReleaseHistoryRetentionLease handle -> do+        liftIO $ handle `shouldBe` sampleHandle+        record calls "release"+        pure (HistoryRetentionReleased sampleLease)+    GetHistoryRetentionLeaseInventory query -> do+        liftIO $ query `shouldBe` sampleQuery+        record calls "inventory"+        pure (Vector.singleton sampleLease)+    PruneHistoryRetentionLeases cutoff -> do+        liftIO $ cutoff `shouldBe` sampleTime+        record calls "prune"+        pure (HistoryRetentionPruneResult 2 3)+    _ -> error "unexpected Store operation in history retention mock"++record :: (IOE :> es) => IORef [Text] -> Text -> Eff es ()+record calls name = liftIO $ modifyIORef' calls (<> [name])++sampleRequest :: HistoryRetentionLeaseRequest+sampleRequest =+    HistoryRetentionLeaseRequest sampleOwner sampleReason sampleDuration++sampleHandle :: HistoryRetentionLeaseHandle+sampleHandle = HistoryRetentionLeaseHandle sampleId sampleOwner++sampleLease :: HistoryRetentionLease+sampleLease =+    HistoryRetentionLease+        sampleId+        sampleOwner+        sampleReason+        (GlobalPosition 17)+        sampleTime+        sampleTime+        sampleExpiry+        Nothing+        HistoryRetentionLeaseActive++sampleId :: HistoryRetentionLeaseId+sampleId = HistoryRetentionLeaseId UUID.nil++sampleOwner :: HistoryRetentionLeaseOwner+sampleOwner = either (error . show) (\value -> value) (mkHistoryRetentionLeaseOwner "mock-owner")++sampleReason :: HistoryRetentionLeaseReason+sampleReason = either (error . show) (\value -> value) (mkHistoryRetentionLeaseReason "mock-reason")++sampleDuration :: HistoryRetentionLeaseDuration+sampleDuration = either (error . show) (\value -> value) (mkHistoryRetentionLeaseDuration (secondsToDiffTime 60))++sampleQuery :: HistoryRetentionInventoryQuery+sampleQuery =+    HistoryRetentionInventoryQuery $+        either (error . show) (\value -> value) (mkHistoryRetentionInventoryLimit 10)++sampleTime :: UTCTime+sampleTime = UTCTime (fromGregorian 2026 8 13) 0++sampleExpiry :: UTCTime+sampleExpiry = UTCTime (fromGregorian 2026 8 13) (secondsToDiffTime 60)

@@ -3,7 +3,7 @@ module Test.PerformanceStructure (spec) where  import Control.Lens ((^.))-import Control.Monad (unless)+import Control.Monad (forM_, unless) import Data.Aeson (Value (..)) import Data.Aeson qualified as Aeson import Data.Aeson.KeyMap qualified as KeyMap@@ -11,6 +11,7 @@ import Data.Foldable (foldl') import Data.Generics.Labels () import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Int (Int64) import Data.Text (Text) import Data.Text qualified as T import Hasql.Decoders qualified as D@@ -49,6 +50,32 @@                 after <- readIORef checkouts                 result `shouldBe` Right []                 after - before `shouldBe` 0++        it "rejects invalid retention requests before pool checkout" $ do+            checkouts <- newIORef (0 :: Int)+            withObservedStore checkouts $ \_store -> do+                before <- readIORef checkouts+                mkHistoryRetentionLeaseOwner "" `shouldSatisfy` either (const True) (const False)+                mkHistoryRetentionLeaseReason "" `shouldSatisfy` either (const True) (const False)+                mkHistoryRetentionInventoryLimit 0 `shouldSatisfy` either (const True) (const False)+                after <- readIORef checkouts+                after - before `shouldBe` 0++        it "keeps every ordinary statement free of retention coordination" $ do+            let ordinarySql =+                    [ Statement.toSql SQL.appendExpectedVersion+                    , Statement.toSql SQL.appendStreamExists+                    , Statement.toSql SQL.appendNoStream+                    , Statement.toSql SQL.appendAnyVersion+                    , Statement.toSql SQL.readStreamForwardStmt+                    , Statement.toSql SQL.readAllForwardStmt+                    , Statement.toSql SQL.softDeleteStreamStmt+                    , Statement.toSql SQL.undeleteStreamStmt+                    , Statement.toSql SQL.setStreamTruncateBeforeStmt+                    ]+            forM_ ordinarySql $ \sql ->+                T.toLower sql `shouldNotSatisfy` T.isInfixOf "history_retention"+ queryPlanSpec :: Spec queryPlanSpec =     describe "production query plans" $@@ -92,6 +119,16 @@                         [("$1", "ARRAY['00000000-0000-0000-0000-000000000001'::uuid]::uuid[]")]                 expectIndex "ix_dead_letters_event_id" plan +            it "active retention lookup uses ix_history_retention_leases_unreleased_expiry" $ \store -> do+                plan <- explainProductionStatement store activeLeaseLookupStmt []+                expectIndex "ix_history_retention_leases_unreleased_expiry" plan++            it "installs no retention trigger for INSERT or UPDATE" $ \store -> do+                result <- Pool.use (store ^. #pool) (Session.statement () retentionTriggerShapeStmt)+                case result of+                    Left err -> expectationFailure ("could not inspect retention triggers: " <> show err)+                    Right shape -> shape `shouldBe` (6, 0)+ withObservedStore :: IORef Int -> (KirokuStore -> IO ()) -> IO () withObservedStore checkouts =     withTestStoreSettings $ \settings ->@@ -187,6 +224,17 @@     ANALYZE events;     ANALYZE stream_events;     ANALYZE dead_letters;+    INSERT INTO history_retention_leases+      (owner, reason, protected_through, created_at, renewed_at, expires_at, released_at)+    SELECT 'performance-owner',+           'performance-reason',+           0,+           clock_timestamp() - interval '1 day',+           clock_timestamp() - interval '1 day',+           clock_timestamp() + interval '1 hour',+           CASE WHEN n <= 9900 THEN clock_timestamp() ELSE NULL END+    FROM generate_series(1, 10000) AS n;+    ANALYZE history_retention_leases;     """  explainProductionStatement ::@@ -264,3 +312,38 @@                 <> show facts                 <> " from:\n"                 <> show plan++activeLeaseLookupStmt :: Statement () ()+activeLeaseLookupStmt =+    Statement.preparable+        """+        SELECT count(*), min(expires_at)+        FROM history_retention_leases+        WHERE released_at IS NULL+          AND expires_at > clock_timestamp()+        HAVING count(*) > 0+        """+        E.noParams+        D.noResult++retentionTriggerShapeStmt :: Statement () (Int64, Int64)+retentionTriggerShapeStmt =+    Statement.preparable+        """+        SELECT count(*),+               count(*) FILTER (WHERE (trigger.tgtype & 4) <> 0 OR (trigger.tgtype & 16) <> 0)+        FROM pg_catalog.pg_trigger AS trigger+        JOIN pg_catalog.pg_class AS relation ON relation.oid = trigger.tgrelid+        JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace+        WHERE namespace.nspname = 'kiroku'+          AND relation.relname IN ('events', 'stream_events', 'streams')+          AND trigger.tgname IN ('protect_replay_history_delete', 'protect_replay_history_truncate')+          AND NOT trigger.tgisinternal+        """+        E.noParams+        ( D.singleRow+            ( (,)+                <$> D.column (D.nonNullable D.int8)+                <*> D.column (D.nonNullable D.int8)+            )+        )

@@ -0,0 +1,205 @@+{-# LANGUAGE MultilineStrings #-}+{-# LANGUAGE NumericUnderscores #-}++module Test.StreamHistoryGuard (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Lens ((^.))+import Control.Monad (forM_, when)+import Data.Aeson qualified as Aeson+import Data.Text qualified as Text+import Data.Vector qualified as Vector+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable)+import Hasql.Transaction qualified as Tx+import Kiroku.Store+import Test.Helpers (makeEvent, withTestStore)+import Test.Hspec++spec :: Spec+spec = describe "stream history guard" $ do+    it "returns exact metadata and pages with production cursor semantics" $+        withTestStore $ \store -> do+            let stream = StreamName "guard-metadata"+            Right _ <- runStoreIO store $ appendToStream stream NoStream (events 3)+            Right _ <- runStoreIO store $ setStreamTruncateBefore stream (StreamVersion 2)+            result <- runTx store $ do+                locked <- lockStreamHistoryForReplayTx stream+                page <- readStreamForwardTx stream (StreamVersion 0) 10+                pure (locked, page)+            case result of+                (Right info, page) -> do+                    info ^. #version `shouldBe` StreamVersion 3+                    info ^. #truncateBefore `shouldBe` StreamVersion 2+                    info ^. #deletedAt `shouldBe` Nothing+                    fmap (^. #streamVersion) (Vector.toList page)+                        `shouldBe` [StreamVersion 2, StreamVersion 3]+                (Left unavailable, _) -> expectationFailure ("guard unexpectedly unavailable: " <> show unavailable)++    it "returns typed unavailable results for missing and reserved streams" $+        withTestStore $ \store -> do+            missing <- runTx store (lockStreamHistoryForReplayTx (StreamName "guard-missing"))+            missing `shouldBe` Left (StreamHistoryNotFound (StreamName "guard-missing"))+            reserved <- runTx store (lockStreamHistoryForReplayTx (StreamName "$all"))+            reserved `shouldBe` Left (StreamHistoryReserved (StreamName "$all"))++    it "returns soft-deleted metadata and blocks undelete until guard completion" $+        withTestStore $ \store -> do+            let stream = StreamName "guard-undelete"+            Right _ <- runStoreIO store $ appendToStream stream NoStream (events 1)+            Right _ <- runStoreIO store $ softDeleteStream stream+            assertBlockedByGuard store stream $ do+                info <- runTx store (lockStreamHistoryForReplayTx stream)+                info `shouldSatisfy` \case Right StreamInfo{deletedAt = Just _} -> True; _ -> False+                runStoreIO store (undeleteStream stream)++    it "blocks append until guard completion" $+        withTestStore $ \store -> do+            let stream = StreamName "guard-append"+            Right _ <- runStoreIO store $ appendToStream stream NoStream (events 1)+            assertBlockedByGuard store stream $+                runStoreIO store (appendToStream stream AnyVersion (events 1))++    it "releases the guard on rollback and lets the blocked append complete" $+        withTestStore $ \store -> do+            let stream = StreamName "guard-rollback"+            Right _ <- runStoreIO store $ appendToStream stream NoStream (events 1)+            assertBlockedByGuardEnding True store stream $+                runStoreIO store (appendToStream stream AnyVersion (events 1))++    it "blocks link into the guarded stream until guard completion" $+        withTestStore $ \store -> do+            let source = StreamName "guard-link-source"+                target = StreamName "guard-link-target"+            Right _ <- runStoreIO store $ appendToStream source NoStream (events 1)+            Right _ <- runStoreIO store $ appendToStream target NoStream (events 1)+            Right sourceRows <- runStoreIO store $ readStreamForward source (StreamVersion 0) 10+            let eventId = Vector.head sourceRows ^. #eventId+            assertBlockedByGuard store target $+                runStoreIO store (linkToStream target [eventId])++    it "blocks soft delete and logical truncate until guard completion" $+        withTestStore $ \store -> do+            let soft = StreamName "guard-soft-delete"+                truncated = StreamName "guard-truncate"+            Right _ <- runStoreIO store $ appendToStream soft NoStream (events 1)+            Right _ <- runStoreIO store $ appendToStream truncated NoStream (events 2)+            assertBlockedByGuard store soft $+                runStoreIO store (softDeleteStream soft)+            assertBlockedByGuard store truncated $+                runStoreIO store (setStreamTruncateBefore truncated (StreamVersion 2))++    it "blocks hard delete of the guarded origin until guard completion" $+        withTestStore $ \store -> do+            let stream = StreamName "guard-hard-delete"+            Right _ <- runStoreIO store $ appendToStream stream NoStream (events 1)+            assertBlockedByGuard store stream $+                runStoreIO store (hardDeleteStream stream)++    it "blocks hard delete of another origin linked into the guarded stream" $+        withTestStore $ \store -> do+            let origin = StreamName "guard-linked-origin"+                guarded = StreamName "guard-linked-target"+            Right _ <- runStoreIO store $ appendToStream origin NoStream (events 1)+            Right _ <- runStoreIO store $ appendToStream guarded NoStream (events 1)+            Right sourceRows <- runStoreIO store $ readStreamForward origin (StreamVersion 0) 10+            Right _ <- runStoreIO store $ linkToStream guarded [Vector.head sourceRows ^. #eventId]+            assertBlockedByGuard store guarded $+                runStoreIO store (hardDeleteStream origin)++    it "keeps opposing hard deletes and multi-stream appends deadlock-free" $+        withTestStore $ \store ->+            forM_ [1 .. 5 :: Int] $ \index -> do+                let suffix = Text.pack (show index)+                    first = StreamName ("guard-race-a-" <> suffix)+                    second = StreamName ("guard-race-b-" <> suffix)+                Right _ <- runStoreIO store $ appendToStream first NoStream (events 1)+                Right _ <- runStoreIO store $ appendToStream second NoStream (events 1)+                operations <-+                    Async.async $+                        Async.concurrently+                            (runStoreIO store $ hardDeleteStream first)+                            ( runStoreIO store $+                                appendMultiStream+                                    [ (first, AnyVersion, events 1)+                                    , (second, AnyVersion, events 1)+                                    ]+                            )+                _ <- waitWithin "hard-delete/multi-append race" operations+                pure ()++assertBlockedByGuard :: KirokuStore -> StreamName -> IO result -> IO ()+assertBlockedByGuard = assertBlockedByGuardEnding False++assertBlockedByGuardEnding :: Bool -> KirokuStore -> StreamName -> IO result -> IO ()+assertBlockedByGuardEnding rollBack store stream mutation = do+    guard <- Async.async $ runStoreIO store $ runTransaction $ do+        locked <- lockStreamHistoryForReplayTx stream+        _ <- Tx.statement () holdGuardStmt+        when rollBack Tx.condemn+        pure locked+    waitForGuardPhase store 100+    waiter <- Async.async mutation+    threadDelay 50_000+    Async.poll waiter >>= \case+        Nothing -> pure ()+        Just _ -> expectationFailure "mutation completed while the stream-history guard was held"+    guardResult <- waitWithin "guard transaction" guard+    guardResult `shouldSatisfy` \case Right (Right _) -> True; _ -> False+    _ <- waitWithin "blocked mutation" waiter+    pure ()++waitForGuardPhase :: KirokuStore -> Int -> IO ()+waitForGuardPhase _ 0 = expectationFailure "guard never reached its held phase"+waitForGuardPhase store attempts = do+    result <- Pool.use (store ^. #pool) (Session.statement () guardActiveStmt)+    case result of+        Right True -> pure ()+        Right False -> threadDelay 10_000 >> waitForGuardPhase store (attempts - 1)+        Left err -> expectationFailure ("could not observe guard phase: " <> show err)++waitWithin :: String -> Async.Async value -> IO value+waitWithin label action = do+    result <- Async.race (threadDelay 2_000_000) (Async.wait action)+    case result of+        Left () -> do+            Async.cancel action+            expectationFailure (label <> " timed out")+            error "unreachable"+        Right value -> pure value++runTx :: KirokuStore -> Tx.Transaction value -> IO value+runTx store transaction = do+    result <- runStoreIO store (runTransaction transaction)+    case result of+        Left err -> expectationFailure ("guard transaction failed: " <> show err) >> error "unreachable"+        Right value -> pure value++events :: Int -> [EventData]+events count =+    [makeEvent "Guarded" (Aeson.object [("n", Aeson.toJSON n)]) | n <- [1 .. count]]++holdGuardStmt :: Statement () Bool+holdGuardStmt =+    preparable+        "SELECT pg_sleep(0.4) IS NULL"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))++guardActiveStmt :: Statement () Bool+guardActiveStmt =+    preparable+        """+        SELECT EXISTS (+          SELECT 1+          FROM pg_stat_activity+          WHERE state = 'active'+            AND query = 'SELECT pg_sleep(0.4) IS NULL'+        )+        """+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))

@@ -1,3 +1,4 @@+{-# LANGUAGE MultilineStrings #-} {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE TypeApplications #-} @@ -11,12 +12,15 @@ import Data.Map.Strict qualified as Map import Data.Text (Text) import Data.Text qualified as T-import Data.Time.Clock (getCurrentTime)+import Data.Time.Clock (UTCTime, getCurrentTime) import Data.Vector qualified as V import Effectful (runEff) import Effectful.Error.Static (runErrorNoCallStack)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable) import Kiroku.Store import Kiroku.Store.SQL qualified as SQL import System.Timeout (timeout)@@ -27,9 +31,10 @@ spec = describe "SubscriptionCheckpointInventory" $ do     it "returns position zero and no rows for an empty migrated store" $         withTestStore $ \store -> do-            SubscriptionCheckpointInventory captured rows <- readInventory store+            inventory@(SubscriptionCheckpointInventory captured rows) <- readInventory store             captured `shouldBe` GlobalPosition 0             rows `shouldBe` V.empty+            assertSqlRelationMatchesInventory store inventory      it "runs through the resource-backed Store interpreter" $         withTestStore $ \store -> do@@ -61,13 +66,14 @@             saveCheckpoint store "alpha" 10 3             saveCheckpoint store "alpha" 2 5 -            SubscriptionCheckpointInventory captured rows <- readInventory store+            inventory@(SubscriptionCheckpointInventory captured rows) <- readInventory store             captured `shouldBe` GlobalPosition 20             checkpointKeys rows                 `shouldBe` [ ("alpha", 2, 5)                            , ("alpha", 10, 3)                            , ("zeta", 2, 7)                            ]+            assertSqlRelationMatchesInventory store inventory      it "preserves monotonic checkpoints and observes later commits on a fresh read" $         withTestStore $ \store -> do@@ -94,6 +100,7 @@             Map.member (name, 0) states `shouldBe` False             inventory <- readInventory store             inventoryKeys inventory `shouldBe` [("stopped", 0, 1)]+            assertSqlRelationMatchesInventory store inventory      it "does not expose in-flight live handler progress before checkpoint commit" $ do         caughtUp <- newEmptyMVar@@ -117,11 +124,13 @@             waitForMVar "live handler did not receive the event" enteredHandler             beforeCommit <- readInventory store             inventoryKeys beforeCommit `shouldBe` [("in-flight", 0, 1)]+            assertSqlRelationMatchesInventory store beforeCommit              putMVar releaseHandler ()             waitClean handle             afterCommit <- readInventory store             inventoryKeys afterCommit `shouldBe` [("in-flight", 0, 2)]+            assertSqlRelationMatchesInventory store afterCommit      it "observes the checkpoint advanced by a dead-letter transaction" $         withTestStore $ \store -> do@@ -196,3 +205,40 @@     [ (name, member, position)     | SubscriptionCheckpoint (SubscriptionName name) member (GlobalPosition position) _ <- V.toList rows     ]++assertSqlRelationMatchesInventory :: KirokuStore -> SubscriptionCheckpointInventory -> Expectation+assertSqlRelationMatchesInventory store (SubscriptionCheckpointInventory _ inventoryRows) = do+    result <- Pool.use (store ^. #pool) $ Session.statement () subscriptionCheckpointRelationStmt+    case result of+        Left err -> expectationFailure ("subscription checkpoint relation failed: " <> show err)+        Right sqlRows ->+            sqlRows+                `shouldBe` [ (name, member, position, updatedAt)+                           | SubscriptionCheckpoint+                                (SubscriptionName name)+                                member+                                (GlobalPosition position)+                                updatedAt <-+                                V.toList inventoryRows+                           ]++subscriptionCheckpointRelationStmt :: Statement () [(Text, Int32, Int64, UTCTime)]+subscriptionCheckpointRelationStmt =+    preparable+        """+        SELECT subscription_name,+               consumer_group_member,+               checkpoint_position,+               checkpoint_updated_at+        FROM kiroku.subscription_checkpoints_v1+        ORDER BY subscription_name, consumer_group_member+        """+        E.noParams+        ( D.rowList+            ( (,,,)+                <$> D.column (D.nonNullable D.text)+                <*> D.column (D.nonNullable D.int4)+                <*> D.column (D.nonNullable D.int8)+                <*> D.column (D.nonNullable D.timestamptz)+            )+        )