diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/kiroku-store.cabal b/kiroku-store.cabal --- a/kiroku-store.cabal +++ b/kiroku-store.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-store -version: 0.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 diff --git a/src/Kiroku/Store.hs b/src/Kiroku/Store.hs --- a/src/Kiroku/Store.hs +++ b/src/Kiroku/Store.hs @@ -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 (..)) diff --git a/src/Kiroku/Store/Effect.hs b/src/Kiroku/Store/Effect.hs --- a/src/Kiroku/Store/Effect.hs +++ b/src/Kiroku/Store/Effect.hs @@ -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 -> diff --git a/src/Kiroku/Store/Error.hs b/src/Kiroku/Store/Error.hs --- a/src/Kiroku/Store/Error.hs +++ b/src/Kiroku/Store/Error.hs @@ -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 diff --git a/src/Kiroku/Store/HistoryRetention.hs b/src/Kiroku/Store/HistoryRetention.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/HistoryRetention.hs @@ -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 diff --git a/src/Kiroku/Store/HistoryRetention/Internal.hs b/src/Kiroku/Store/HistoryRetention/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/HistoryRetention/Internal.hs @@ -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 diff --git a/src/Kiroku/Store/HistoryRetention/SQL.hs b/src/Kiroku/Store/HistoryRetention/SQL.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/HistoryRetention/SQL.hs @@ -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) diff --git a/src/Kiroku/Store/HistoryRetention/Types.hs b/src/Kiroku/Store/HistoryRetention/Types.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/HistoryRetention/Types.hs @@ -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 diff --git a/src/Kiroku/Store/Lifecycle.hs b/src/Kiroku/Store/Lifecycle.hs --- a/src/Kiroku/Store/Lifecycle.hs +++ b/src/Kiroku/Store/Lifecycle.hs @@ -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 :: diff --git a/src/Kiroku/Store/Observability.hs b/src/Kiroku/Store/Observability.hs --- a/src/Kiroku/Store/Observability.hs +++ b/src/Kiroku/Store/Observability.hs @@ -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 diff --git a/src/Kiroku/Store/Transaction.hs b/src/Kiroku/Store/Transaction.hs --- a/src/Kiroku/Store/Transaction.hs +++ b/src/Kiroku/Store/Transaction.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/Test/HistoryRetention.hs b/test/Test/HistoryRetention.hs new file mode 100644 --- /dev/null +++ b/test/Test/HistoryRetention.hs @@ -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))) diff --git a/test/Test/HistoryRetentionMock.hs b/test/Test/HistoryRetentionMock.hs new file mode 100644 --- /dev/null +++ b/test/Test/HistoryRetentionMock.hs @@ -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) diff --git a/test/Test/PerformanceStructure.hs b/test/Test/PerformanceStructure.hs --- a/test/Test/PerformanceStructure.hs +++ b/test/Test/PerformanceStructure.hs @@ -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) + ) + ) diff --git a/test/Test/StreamHistoryGuard.hs b/test/Test/StreamHistoryGuard.hs new file mode 100644 --- /dev/null +++ b/test/Test/StreamHistoryGuard.hs @@ -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))) diff --git a/test/Test/SubscriptionCheckpointInventory.hs b/test/Test/SubscriptionCheckpointInventory.hs --- a/test/Test/SubscriptionCheckpointInventory.hs +++ b/test/Test/SubscriptionCheckpointInventory.hs @@ -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) + ) + )