kiroku-store 0.12.0.0 → 0.13.0.0

raw patch · 29 files changed

+3024/−11 lines, 29 filesdep +cryptondep +ramdep ~aesondep ~basedep ~generic-lensPVP ok

version bump matches the API change (PVP)

Dependencies added: crypton, ram

Dependency ranges changed: aeson, base, generic-lens, hasql, hasql-pool, lens, text, vector

API changes (from Hackage documentation)

Files

• CHANGELOG.md +44/−0
• bench/CompactionCost.hs +77/−0
• kiroku-store.cabal +35/−1
• src/Kiroku/Store.hs +2/−0
• src/Kiroku/Store/Compaction.hs +74/−0
• src/Kiroku/Store/Compaction/Internal.hs +264/−0
• src/Kiroku/Store/Compaction/SQL.hs +185/−0
• src/Kiroku/Store/Compaction/Types.hs +672/−0
• src/Kiroku/Store/Effect.hs +122/−2
• src/Kiroku/Store/Error.hs +8/−0
• src/Kiroku/Store/Observability.hs +10/−1
• src/Kiroku/Store/Read.hs +31/−0
• src/Kiroku/Store/SQL.hs +80/−0
• src/Kiroku/Store/Transaction.hs +33/−1
• src/Kiroku/Store/Types.hs +42/−1
• test/Main.hs +18/−0
• test/Test/CompactionApply.hs +395/−0
• test/Test/CompactionApplyMock.hs +41/−0
• test/Test/CompactionHelpers.hs +58/−0
• test/Test/CompactionManifest.hs +235/−0
• test/Test/CompactionPreview.hs +214/−0
• test/Test/CompactionPreviewMock.hs +35/−0
• test/Test/EventReferences.hs +116/−0
• test/Test/EventReferencesMock.hs +35/−0
• test/Test/NotifyGuard.hs +6/−0
• test/Test/PerformanceStructure.hs +114/−2
• test/Test/StoreIdentity.hs +43/−0
• test/Test/StoreIdentityMock.hs +32/−0
• test/Test/SubscriptionDeadLetters.hs +3/−3

@@ -1,5 +1,49 @@ # Changelog +## 0.13.0.0 — 2026-10-28++### Breaking Changes++- The closed `Store` effect gains `GetStoreIdentity`, `LookupEventReferences`,+  `PreviewCompaction`, `ApplyCompaction` and `GetCompactionLedger`; exhaustive+  custom interpreters must add arms.+- `StoreError` gains retryable `CompactionConcurrentMutation`. An affected apply+  transaction rolled back; preview again before retrying. Direct transaction+  composers can also receive existing `UnexpectedServerError` SQLSTATE `23503`+  for a concurrent foreign-key reference and must treat it identically.+- Exhaustive `KirokuEvent` observers must handle `KirokuEventCompactionPreviewed`,+  `KirokuEventCompactionRefused`, `KirokuEventCompactionApplied` and+  `KirokuEventCompactionAlreadyApplied`.++### New Features++- Add `Kiroku.Store.Compaction`, `.Types` and `.SQL`: immutable digest-sealed+  manifests, explicit reference policies, typed refusals, canonical SHA-256+  encodings, deterministic reports and durable ledger records. JSON decoding+  verifies manifest and report digests.+- Add read-only `previewCompaction` / `previewCompactionTx`. The effect uses one+  pooled connection and one repeatable-read snapshot without row locks or writes.+- Add atomic `applyCompaction` / `applyCompactionTx`: revalidate exact witnesses+  under the retention coordinator then ascending stream-ID locks, refuse active+  replay-history leases, and delete only selected events and acknowledged links.+  Retained envelopes, versions, positions and stream heads stay unchanged; ordinary+  append/read SQL and triggers gain no work. Causation refusal is validation-time+  best-effort. Reference/count races roll back the whole transaction.+- Add `compactionLedger` / `compactionLedgerTx`. The append-only record commits+  with deletion; an identical reapply returns that original record. Keep the+  reviewed manifest and archived envelopes because the ledger omits event IDs.+- Add `StoreIdentity`, `storeIdentity` and `storeIdentityTx`. Restored clones share+  their source identity. These APIs and compaction require migration `0017` from+  `kiroku-store-migrations` 0.7.3.0; older schemas report a missing relation.+- Add `lookupEventReferences` / `lookupEventReferencesTx`: home/global/link+  memberships, dead-letter counts and causation dependents outside the requested+  set. Unknown IDs are omitted, duplicates deduplicated and empty input uses no+  connection. Membership and dead-letter reads batch at 10,000 IDs on one checkout;+  causation exclusions cover the complete request.+- Document the reviewed workflow, lease margins, 10k initial operational chunks,+  audit and separate PostgreSQL space reclamation in the+  [operator guide](https://github.com/shinzui/kiroku/blob/master/docs/user/compaction.md).+ ## 0.12.0.0 — 2026-10-26  ### Breaking Changes

@@ -0,0 +1,77 @@+{-# LANGUAGE OverloadedRecordDot #-}++-- A bounded maintenance-path sizing probe, not a statistical regression gate.+module Main (main) where++import Control.Lens ((^.))+import Control.Monad (forM_, unless)+import Data.Aeson (object, (.=))+import Data.Generics.Labels ()+import Data.Text qualified as T+import Data.Vector qualified as V+import GHC.Clock (getMonotonicTimeNSec)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Kiroku.Store hiding (id)+import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres)+import Text.Printf (printf)++main :: IO ()+main = withSharedMigratedPostgres $ do+  putStrLn "selections,link_memberships,dead_letters,apply_wall_seconds,coordinator_hold_upper_bound_seconds,microseconds_per_selection,report_digest"+  forM_ [10000, 50000, 100000 :: Int] $ \size ->+    withMigratedTestDatabase $ \conn -> withStore (defaultConnectionSettings conn) $ \store -> do+      let origin = StreamName "cost-origin"+          target = StreamName "cost-links"+      forM_ [0, 2500 .. size - 1] $ \offset -> do+        let events =+              [ EventData+                  Nothing+                  (EventType "CompactionCost")+                  (object ["ordinal" .= n, "payload" .= T.replicate 256 "x"])+                  Nothing+                  Nothing+                  Nothing+              | n <- [offset + 1 .. min size (offset + 2500)]+              ]+        _ <- require =<< runStoreIO store (appendToStream origin AnyVersion events)+        pure ()+      events <- require =<< runStoreIO store (readAllForward (GlobalPosition 0) (fromIntegral size))+      let linked = V.filter (\e -> let GlobalPosition p = e.globalPosition in p `mod` 10 == 0) events+      _ <- require =<< runStoreIO store (linkToStream target (map (\e -> e.eventId) (V.toList linked)))+      require+        =<< Pool.use+          (store ^. #pool)+          ( Session.script+              "INSERT INTO dead_letters(subscription_name,consumer_group_member,global_position,event_id,reason,reason_summary,attempt_count) SELECT 'cost',0,stream_version,event_id,'{}'::jsonb,'cost fixture',1 FROM stream_events WHERE stream_id=0 AND stream_version % 10=0"+          )+      identity <- require =<< runStoreIO store storeIdentity+      op <- either (fail . show) pure (mkCompactionOperation ("cost-" <> T.pack (show size)))+      let selections =+            [ CompactionSelection+                e.eventId+                origin+                e.originalVersion+                e.globalPosition+                (let GlobalPosition pos = e.globalPosition in if pos `mod` 10 == 0 then V.singleton (LinkWitness target (StreamVersion (pos `div` 10))) else V.empty)+            | e <- V.toList events+            ]+          heads = [StreamHeadWitness origin (StreamVersion (fromIntegral size)), StreamHeadWitness target (StreamVersion (fromIntegral (size `div` 10)))]+      manifest <- either (fail . show) pure (mkCompactionManifest (CompactionManifestInput identity op RemoveDeadLetters RefuseCausationDependents heads selections Nothing))+      start <- getMonotonicTimeNSec+      result <- require =<< runStoreIO store (applyCompaction manifest)+      end <- getMonotonicTimeNSec+      record <- case result of+        Right (CompactionAppliedNow value) -> pure value+        other -> fail (show other)+      let seconds = fromIntegral (end - start) / 1e9 :: Double+          report = record.report+      unless (report.selectedEvents == fromIntegral size && report.linkMemberships == fromIntegral (size `div` 10) && report.deadLettersRemoved == fromIntegral (size `div` 10)) (fail "count mismatch")+      ledger <- require =<< runStoreIO store (compactionLedger (CompactionLedgerQuery (either (error . show) id (mkCompactionLedgerLimit 10))))+      unless (ledger == V.singleton record && report.reportDigest == compactionReportDigest report) (fail "ledger mismatch")+      repeated <- require =<< runStoreIO store (applyCompaction manifest)+      unless (repeated == Right (CompactionAlreadyApplied record)) (fail "reapply mismatch")+      printf "%d,%d,%d,%.6f,%.6f,%.3f,%s\n" size (size `div` 10) (size `div` 10) seconds seconds (seconds * 1e6 / fromIntegral size) (T.unpack (compactionDigestHex report.reportDigest))++require :: (Show e) => Either e a -> IO a+require = either (fail . show) pure

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.12.0.0+version:         0.13.0.0 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -38,6 +38,9 @@     Kiroku.Store     Kiroku.Store.Append     Kiroku.Store.Causation+    Kiroku.Store.Compaction+    Kiroku.Store.Compaction.SQL+    Kiroku.Store.Compaction.Types     Kiroku.Store.Connection     Kiroku.Store.Effect     Kiroku.Store.Effect.Resource@@ -63,6 +66,7 @@     Kiroku.Store.Types    other-modules:+    Kiroku.Store.Compaction.Internal     Kiroku.Store.HistoryRetention.Internal     Kiroku.Store.HistoryRetention.SQL     Kiroku.Store.Subscription.Checkpoint.SQL@@ -76,6 +80,7 @@     , bytestring            >=0.11  && <0.13     , containers            >=0.6   && <0.8     , contravariant-extras  >=0.3   && <0.4+    , crypton               >=1.1.5 && <2.3     , effectful-core        >=2.6.1 && <2.7  || >=2.7.1.1 && <2.8     , generic-lens          >=2.2   && <2.4     , hasql                 >=1.10  && <1.11@@ -85,6 +90,7 @@     , lens                  >=5.2   && <5.4     , mmzk-typeid           >=0.6   && <0.8     , mtl                   >=2.3   && <2.4+    , ram                   >=0.22  && <0.23     , stm                   >=2.5   && <2.6     , streamly-core         >=0.3   && <0.4     , text                  >=2.0   && <2.2@@ -109,12 +115,20 @@     Test.Category     Test.CategoryIdleNoSpin     Test.Causation+    Test.CompactionApply+    Test.CompactionApplyMock+    Test.CompactionHelpers+    Test.CompactionManifest+    Test.CompactionPreview+    Test.CompactionPreviewMock     Test.Concurrency     Test.ConsumerGroup     Test.ConsumerGroupEffect     Test.ConsumerGroupResize     Test.ConsumerGroupSql     Test.DeadLetterQueryPlans+    Test.EventReferences+    Test.EventReferencesMock     Test.EventTypeFilter     Test.FailureInjection     Test.HandlerStall@@ -133,6 +147,8 @@     Test.ReadStream     Test.ScriptSessionErrors     Test.StartupFailureSurfacing+    Test.StoreIdentity+    Test.StoreIdentityMock     Test.StreamBridgeTermination     Test.StreamHead     Test.StreamHeadIsolation@@ -361,3 +377,21 @@     , text                 >=2.0   && <2.2     , time                 >=1.12  && <1.15     , vector               >=0.13  && <0.14++benchmark kiroku-compaction-cost+  import:         common+  type:           exitcode-stdio-1.0+  main-is:        CompactionCost.hs+  hs-source-dirs: bench+  ghc-options:    -threaded -rtsopts "-with-rtsopts=-N2 -A32m"+  build-depends:+    , aeson+    , base+    , generic-lens+    , hasql+    , hasql-pool+    , kiroku-store+    , kiroku-test-support+    , lens+    , text+    , vector

@@ -8,6 +8,7 @@ -- explicitly to use the effectful version. module Kiroku.Store   ( module Kiroku.Store.Types,+    module Kiroku.Store.Compaction,     module Kiroku.Store.Connection,     module Kiroku.Store.Effect,     module Kiroku.Store.Effect.Resource,@@ -52,6 +53,7 @@ import Hasql.Pool.Observation (ConnectionReadyForUseReason (..), ConnectionStatus (..), ConnectionTerminationReason (..), Observation (..)) import Kiroku.Store.Append import Kiroku.Store.Causation+import Kiroku.Store.Compaction import Kiroku.Store.Connection import Kiroku.Store.Effect import Kiroku.Store.Effect.Resource

@@ -0,0 +1,74 @@+-- | Explicit selective physical compaction. Preview is read-only, takes no row+-- locks, and reports discrepancies under one snapshot. Apply revalidates under+-- the retention coordinator and ordered stream locks before deleting anything.+-- Unlike logical truncate-before, compaction physically removes selected events;+-- unlike whole-stream hard delete, it keeps streams live and never renumbers+-- retained history. PostgreSQL vacuum/rewrite work controls space reclamation;+-- apply does not shrink relation files. See @docs/user/compaction.md@.+module Kiroku.Store.Compaction+  ( module Kiroku.Store.Compaction.Types,+    previewCompaction,+    previewCompactionTx,+    applyCompaction,+    applyCompactionTx,+    compactionLedger,+    compactionLedgerTx,++    -- * Failure-injection seam+    applyCompactionTxWith,+  )+where++import Data.List.NonEmpty (NonEmpty)+import Data.Vector (Vector)+import Effectful (Eff, (:>))+import Effectful.Dispatch.Dynamic (send)+import Hasql.Transaction qualified as Tx+import Kiroku.Store.Compaction.Internal qualified as Internal+import Kiroku.Store.Compaction.Types+import Kiroku.Store.Effect (Store (..))++-- | Read-only preview on one pooled connection and one repeatable-read snapshot.+-- Reports active leases without waiting for them. Causation refusal is a+-- validation-time check: causation IDs have no foreign key and dependents may+-- be appended to unlisted streams after validation.+previewCompaction :: (Store :> es) => CompactionManifest -> Eff es (Either (NonEmpty CompactionRefusal) CompactionPreview)+previewCompaction = send . PreviewCompaction++-- | Preview inside the caller's transaction. Takes no row locks and writes+-- nothing. The caller chooses isolation; use repeatable read for one snapshot.+previewCompactionTx :: CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionPreview)+previewCompactionTx = Internal.previewCompactionTx++-- | Apply one manifest atomically. Takes the retention coordinator first,+-- refuses active leases, recognizes recorded digests before locking streams,+-- then locks affected streams by ascending internal ID and revalidates every+-- witness before deletion. Retained rows and stream heads are unchanged; no+-- NOTIFY fires. A repeated recorded manifest returns its original ledger record.+-- Concurrent references abort the whole transaction: re-preview and re-apply.+applyCompaction :: (Store :> es) => CompactionManifest -> Eff es (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+applyCompaction = send . ApplyCompaction++-- | Apply inside the caller's write transaction. Lease work must precede stream+-- guards to preserve the coordinator-before-stream lock order. Count races+-- raise 'Kiroku.Store.Error.CompactionConcurrentMutation'; a concurrent foreign+-- key reference may surface as 'Kiroku.Store.Error.UnexpectedServerError' with+-- SQLSTATE @23503@ for direct composers. Both mean rollback and retry after a+-- fresh preview. Causation refusal is validation-time best-effort.+applyCompactionTx :: CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+applyCompactionTx = Internal.applyCompactionTx++-- | Newest durable audit records first, with a validated limit of 1–1000.+-- Records preserve stored digests verbatim for audit; the ledger omits event IDs.+compactionLedger :: (Store :> es) => CompactionLedgerQuery -> Eff es (Vector CompactionRecord)+compactionLedger = send . GetCompactionLedger++-- | Ledger inventory inside the caller's transaction.+compactionLedgerTx :: CompactionLedgerQuery -> Tx.Transaction (Vector CompactionRecord)+compactionLedgerTx = Internal.compactionLedgerTx++-- | Failure-injection barrier after validation and before the first DELETE.+-- Production 'applyCompactionTx' uses @pure ()@ and adds no statement. A barrier+-- must remain SQL-only when the caller enables automatic transaction retries.+applyCompactionTxWith :: Tx.Transaction () -> CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+applyCompactionTxWith = Internal.applyCompactionTxWith

@@ -0,0 +1,264 @@+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE NoFieldSelectors #-}++module Kiroku.Store.Compaction.Internal+  ( ValidatedCompaction (..),+    ValidationOutcome (..),+    LedgerOutcome (..),+    ledgerOutcomeTx,+    validateCompactionTx,+    buildReport,+    previewCompactionTx,+    applyCompactionTx,+    applyCompactionTxWith,+    compactionLedgerTx,+  )+where++import Control.Monad (unless, void)+import Data.Int (Int64)+import Data.List (find)+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Map.Strict qualified as Map+import Data.Set qualified as Set+import Data.Text.Encoding qualified as TE+import Data.UUID (UUID)+import Data.Vector (Vector)+import Data.Vector qualified as V+import GHC.Generics (Generic)+import Hasql.Transaction qualified as Tx+import Kiroku.Store.Compaction.SQL qualified as SQL+import Kiroku.Store.Compaction.Types+import Kiroku.Store.HistoryRetention.Internal qualified as Retention+import Kiroku.Store.SQL qualified as StoreSQL+import Kiroku.Store.Types++data ValidatedCompaction = ValidatedCompaction+  { report :: !CompactionReport,+    selectedEventIds :: !(Vector UUID),+    junctionRows :: !(Vector (UUID, Int64))+  }+  deriving stock (Eq, Show, Generic)++data ValidationOutcome+  = ValidationRefused (NonEmpty CompactionRefusal)+  | ValidationAlreadyApplied CompactionRecord+  | ValidationReady ValidatedCompaction+  deriving stock (Eq, Show)++data LedgerOutcome = LedgerUnrecorded | LedgerAlreadyApplied CompactionRecord | LedgerConflict CompactionId Int64+  deriving stock (Eq, Show)++selectedIds :: CompactionManifest -> Vector UUID+selectedIds = V.map (\s -> let EventId uid = s.eventId in uid) . manifestSelections++ledgerOutcomeTx :: CompactionManifest -> Tx.Transaction LedgerOutcome+ledgerOutcomeTx manifest = do+  found <- Tx.statement (compactionDigestBytes (compactionManifestDigest manifest)) SQL.ledgerRecordByDigestStmt+  case found of+    Nothing -> pure LedgerUnrecorded+    Just record -> do+      survivors <- Tx.statement (selectedIds manifest) SQL.survivingSelectedEventsStmt+      pure (if survivors == 0 then LedgerAlreadyApplied record else LedgerConflict record.compactionId survivors)++-- No import of Store.Transaction: it imports Effect, which consumes this module.+validateCompactionTx :: CompactionManifest -> Tx.Transaction ValidationOutcome+validateCompactionTx manifest = do+  identity <- Tx.statement () StoreSQL.storeIdentityStmt+  if identity /= manifestStoreIdentity manifest+    then pure (ValidationRefused (CompactionStoreIdentityMismatch (manifestStoreIdentity manifest) identity :| []))+    else do+      conflict <- Retention.activeHistoryRetentionConflictTx+      let leaseRefusals = maybe [] (pure . CompactionHistoryRetentionActive) conflict+      ledger <- ledgerOutcomeTx manifest+      case ledger of+        LedgerAlreadyApplied record -> pure $ case NE.nonEmpty leaseRefusals of+          Just refusals -> ValidationRefused refusals+          Nothing -> ValidationAlreadyApplied record+        LedgerConflict cid count -> pure (ValidationRefused (CompactionLedgerConflict cid count :| []))+        LedgerUnrecorded -> validateWitnessesTx manifest leaseRefusals++validateWitnessesTx :: CompactionManifest -> [CompactionRefusal] -> Tx.Transaction ValidationOutcome+validateWitnessesTx manifest leaseRefusals = do+  streams <- Tx.statement (manifestStreamNames manifest) SQL.resolveStreamsStmt+  let resolved = Map.fromList [(StreamName name, (sid, ver, deleted)) | (name, sid, ver, deleted) <- V.toList streams]+      allSelections = V.toList (manifestSelections manifest)+      resolvable = filter (\s -> Map.member s.originStream resolved) allSelections+      ids = V.fromList [uid | s <- resolvable, let EventId uid = s.eventId]+      streamRefusals = concatMap (checkStream resolved) (V.toList (manifestStreamHeads manifest))+  rows <- if V.null ids then pure V.empty else Tx.statement ids SQL.eventMembershipsStmt+  let grouped = Map.map reverse (Map.fromListWith (<>) [(EventId uid, [(sid, StreamName name, ver, origin, originVer)]) | (uid, sid, name, ver, origin, originVer) <- V.toList rows])+      knownNames =+        Map.fromList [(sid, StreamName name) | (name, sid, _, _) <- V.toList streams]+          <> Map.fromList [(sid, StreamName name) | (_, sid, name, _, _, _) <- V.toList rows]+      missingOriginIds = Set.toList (Set.fromList [origin | (_, _, _, _, origin, _) <- V.toList rows, Map.notMember origin knownNames])+  extraNames <- if null missingOriginIds then pure V.empty else Tx.statement missingOriginIds StoreSQL.lookupStreamNamesStmt+  -- The causation exclusion must cover the whole manifest, even when another+  -- selected event has an invalid witness. Report policies only for verified rows.+  deadLetters <- Tx.statement (selectedIds manifest) SQL.deadLetterCountsStmt+  dependents <- Tx.statement (selectedIds manifest) SQL.causationDependentCountsStmt+  let names = knownNames <> Map.fromList [(sid, StreamName name) | (sid, name) <- V.toList extraNames]+      deadMap = Map.fromList (V.toList deadLetters)+      dependentMap = Map.fromList (V.toList dependents)+      results = map (checkSelection resolved grouped names deadMap dependentMap) resolvable+      refusals = leaseRefusals <> streamRefusals <> concatMap fst results+      pairs = concatMap snd results+      links = sum [fromIntegral (V.length s.acknowledgedLinks) | s <- allSelections]+      deadCount = if manifestDeadLetterPolicy manifest == RemoveDeadLetters then sum (Map.elems deadMap) else 0+      causeCount = if manifestCausationPolicy manifest == AllowDanglingCausation then sum (Map.elems dependentMap) else 0+  pure $ case NE.nonEmpty refusals of+    Just found -> ValidationRefused found+    Nothing -> ValidationReady (ValidatedCompaction (buildReport manifest links deadCount causeCount) (selectedIds manifest) (V.fromList pairs))+  where+    checkStream resolved h = case Map.lookup h.stream resolved of+      Nothing -> [CompactionStreamMissing h.stream]+      Just (_, ver, deleted) ->+        [CompactionStreamSoftDeleted h.stream | deleted]+          <> [CompactionStreamHeadDrift h.stream h.headVersion (StreamVersion ver) | StreamVersion ver /= h.headVersion]+    checkSelection resolved grouped names deadMap dependentMap s =+      case Map.findWithDefault [] s.eventId grouped of+        [] -> ([CompactionSelectedEventMissing s.eventId], [])+        liveRows@((_, _, _, actualOrigin, actualVersion) : _) ->+          let EventId uid = s.eventId+              originId = maybe (-1) (\(sid, _, _) -> sid) (Map.lookup s.originStream resolved)+              actualName = Map.findWithDefault s.originStream actualOrigin names+              global = fmap (\(_, _, v, _, _) -> GlobalPosition v) (find (\(sid, _, _, _, _) -> sid == 0) liveRows)+              home = find (\(sid, _, _, _, _) -> sid == originId) liveRows+              matchesHome = case home of+                Just (_, _, ver, origin, originVer) -> origin == originId && StreamVersion ver == s.originVersion && StreamVersion originVer == s.originVersion+                Nothing -> False+              originsMatch = all (\(_, _, _, origin, ver) -> origin == originId && StreamVersion ver == s.originVersion) liveRows+              witnesses =+                [ CompactionWitnessMismatch s.eventId (WitnessMismatch actualName (StreamVersion actualVersion) global)+                | not (matchesHome && originsMatch && global == Just s.globalPosition)+                ]+              actualLinks = [(LinkWitness name (StreamVersion ver), sid) | (sid, name, ver, origin, _) <- liveRows, sid /= 0, sid /= origin]+              expectedLinks = V.toList s.acknowledgedLinks+              key (LinkWitness name ver) = (name, ver)+              actualKeys = Set.fromList (map (key . fst) actualLinks)+              expectedKeys = Set.fromList (map key expectedLinks)+              linkRefusals =+                [CompactionUnexpectedLink s.eventId link | (link, _) <- actualLinks, Set.notMember (key link) expectedKeys]+                  <> [CompactionAcknowledgedLinkMissing s.eventId link | link <- expectedLinks, Map.member link.stream resolved, Set.notMember (key link) actualKeys]+              verified = null witnesses && null linkRefusals+              dead = Map.findWithDefault 0 uid deadMap+              cause = Map.findWithDefault 0 uid dependentMap+              referenceRefusals =+                if not verified+                  then []+                  else+                    [CompactionDeadLettersPresent s.eventId dead | dead > 0, manifestDeadLetterPolicy manifest == RefuseDeadLetters]+                      <> [CompactionCausationDependentsPresent s.eventId cause | cause > 0, manifestCausationPolicy manifest == RefuseCausationDependents]+              pairs = if verified then (uid, originId) : (uid, 0) : [(uid, sid) | (_, sid) <- actualLinks] else []+           in (witnesses <> linkRefusals <> referenceRefusals, pairs)++buildReport :: CompactionManifest -> Int64 -> Int64 -> Int64 -> CompactionReport+buildReport manifest links dead cause =+  sealCompactionReport $+    CompactionReport+      digest+      (manifestStoreIdentity manifest)+      (manifestOperation manifest)+      (manifestDeadLetterPolicy manifest)+      (manifestCausationPolicy manifest)+      count+      count+      count+      links+      dead+      cause+      (V.head selections).globalPosition+      (V.last selections).globalPosition+      (manifestStreamHeads manifest)+      digest+  where+    selections = manifestSelections manifest+    count = fromIntegral (V.length selections)+    digest = compactionManifestDigest manifest++previewCompactionTx :: CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionPreview)+previewCompactionTx manifest =+  validateCompactionTx manifest >>= \case+    ValidationRefused refusals -> pure (Left refusals)+    ValidationAlreadyApplied record -> pure (Right (CompactionPreviewAlreadyApplied record))+    ValidationReady validated -> pure (Right (CompactionPreviewReady validated.report))++applyCompactionTx :: CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+applyCompactionTx = applyCompactionTxWith (pure ())++applyCompactionTxWith :: Tx.Transaction () -> CompactionManifest -> Tx.Transaction (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+applyCompactionTxWith barrier manifest = do+  identity <- Tx.statement () StoreSQL.storeIdentityStmt+  if identity /= manifestStoreIdentity manifest+    then pure (Left (CompactionStoreIdentityMismatch (manifestStoreIdentity manifest) identity :| []))+    else do+      Tx.sql "SET LOCAL kiroku.enable_hard_deletes = 'on'"+      Retention.lockHistoryRetentionCoordinatorTx+      conflict <- Retention.activeHistoryRetentionConflictTx+      case conflict of+        Just active -> pure (Left (CompactionHistoryRetentionActive active :| []))+        Nothing ->+          ledgerOutcomeTx manifest >>= \case+            LedgerAlreadyApplied record -> pure (Right (CompactionAlreadyApplied record))+            LedgerConflict cid count -> pure (Left (CompactionLedgerConflict cid count :| []))+            LedgerUnrecorded -> do+              void (Tx.statement (manifestStreamNames manifest) SQL.lockCompactionStreamsStmt)+              validateCompactionTx manifest >>= \case+                ValidationRefused refusals -> pure (Left refusals)+                ValidationAlreadyApplied record -> pure (Right (CompactionAlreadyApplied record))+                ValidationReady validated -> do+                  barrier+                  Right . CompactionAppliedNow <$> deleteAndRecord validated+  where+    deleteAndRecord validated = do+      let r = validated.report+          (eventKeys, streamKeys) = V.unzip validated.junctionRows+      removed <- Tx.statement (eventKeys, streamKeys) SQL.deleteJunctionRowsStmt+      guardCount "junction count" (fromIntegral (V.length validated.junctionRows)) (fromIntegral (V.length removed))+      let home = V.length (V.filter (\(sid, origin) -> sid /= 0 && sid == origin) removed)+          global = V.length (V.filter (\(sid, _) -> sid == 0) removed)+          links = V.length removed - home - global+      guardCount "home membership count" r.homeMemberships (fromIntegral home)+      guardCount "global membership count" r.globalMemberships (fromIntegral global)+      guardCount "link membership count" r.linkMemberships (fromIntegral links)+      if manifestDeadLetterPolicy manifest == RemoveDeadLetters+        then Tx.statement validated.selectedEventIds SQL.deleteSelectedDeadLettersStmt >>= guardCount "dead-letter count" r.deadLettersRemoved+        else pure ()+      Tx.statement validated.selectedEventIds SQL.deleteSelectedEventsStmt >>= guardCount "events count" r.selectedEvents+      let inserted =+            SQL.CompactionRecordInsert+              r.manifestDigest+              r.reportDigest+              r.storeIdentity+              r.operation+              r.deadLetterPolicy+              r.causationPolicy+              r.selectedEvents+              r.homeMemberships+              r.globalMemberships+              r.linkMemberships+              r.deadLettersRemoved+              r.causationDependents+              r.lowestGlobalPosition+              r.highestGlobalPosition+              r.affectedStreams+      (cid, applied, actor) <- Tx.statement inserted SQL.insertCompactionRecordStmt+      pure (CompactionRecord (CompactionId cid) r applied actor)+    -- Only fixed guard labels and a validated lowercase hex digest enter SQL.+    -- PostgreSQL aborts the transaction itself; returning Left here would commit+    -- earlier deletes in a composer that kept running after the result.+    guardCount label expected actual =+      unless (expected == actual) $+        Tx.sql+          ( TE.encodeUtf8+              ( "DO $$ BEGIN RAISE EXCEPTION USING ERRCODE='KRCMP', MESSAGE='"+                  <> label+                  <> "', DETAIL='"+                  <> compactionDigestHex (compactionManifestDigest manifest)+                  <> "'; END $$"+              )+          )++compactionLedgerTx :: CompactionLedgerQuery -> Tx.Transaction (Vector CompactionRecord)+compactionLedgerTx query = Tx.statement (compactionLedgerLimitValue query.limit) SQL.compactionLedgerStmt

@@ -0,0 +1,185 @@+{-# LANGUAGE ApplicativeDo #-}+{-# LANGUAGE MultilineStrings #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE NoFieldSelectors #-}++-- | Maintenance statements, exposed for deterministic query-plan checks.+module Kiroku.Store.Compaction.SQL+  ( resolveStreamsStmt,+    survivingSelectedEventsStmt,+    ledgerRecordByDigestStmt,+    compactionRecordRow,+    ledgerColumns,+    lockCompactionStreamsStmt,+    deleteJunctionRowsStmt,+    deleteSelectedDeadLettersStmt,+    deleteSelectedEventsStmt,+    CompactionRecordInsert (..),+    insertCompactionRecordStmt,+    compactionLedgerStmt,+    eventMembershipsStmt,+    deadLetterCountsStmt,+    causationDependentCountsStmt,+  )+where++import Contravariant.Extras (contrazip2)+import Data.Aeson qualified as Aeson+import Data.ByteString (ByteString)+import Data.Functor.Contravariant ((>$<))+import Data.Int (Int32, Int64)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time (UTCTime)+import Data.UUID (UUID)+import Data.Vector (Vector)+import GHC.Generics (Generic)+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Kiroku.Store.Compaction.Types+import Kiroku.Store.SQL (causationDependentCountsStmt, deadLetterCountsStmt, eventMembershipsStmt)+import Kiroku.Store.Types (GlobalPosition (..), StoreIdentity (..))++resolveStreamsStmt :: Statement (Vector Text) (Vector (Text, Int64, Int64, Bool))+resolveStreamsStmt =+  preparable+    """+    SELECT s.stream_name, s.stream_id, s.stream_version, s.deleted_at IS NOT NULL+    FROM unnest($1::text[]) AS requested(stream_name)+    JOIN streams s USING (stream_name)+    """+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.text))))+    (D.rowVector ((,,,) <$> col D.text <*> col D.int8 <*> col D.int8 <*> col D.bool))++survivingSelectedEventsStmt :: Statement (Vector UUID) Int64+survivingSelectedEventsStmt =+  preparable+    "SELECT count(*) FROM events WHERE event_id = ANY($1::uuid[])"+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    (D.singleRow (col D.int8))++ledgerColumns :: Text+ledgerColumns = "compaction_id, manifest_digest, report_digest, store_id, operation, dead_letter_policy, causation_policy, selected_events, home_memberships, global_memberships, link_memberships, dead_letters_removed, causation_dependents, lowest_global_position, highest_global_position, affected_streams, applied_at, applied_by"++ledgerRecordByDigestStmt :: Statement ByteString (Maybe CompactionRecord)+ledgerRecordByDigestStmt =+  preparable+    ("SELECT " <> ledgerColumns <> " FROM event_compactions WHERE manifest_digest = $1")+    (E.param (E.nonNullable E.bytea))+    (D.rowMaybe compactionRecordRow)++-- | Decode the stored report digest verbatim. Audits compare it with the+-- recomputed digest; reading an audit record must not silently reseal it.+compactionRecordRow :: D.Row CompactionRecord+compactionRecordRow = do+  cid <- CompactionId <$> col D.uuid+  md <- col digestValue+  rd <- col digestValue+  identity <- StoreIdentity <$> col D.uuid+  operation <- col (D.refine (either (Left . T.pack . show) Right . mkCompactionOperation) D.text)+  dead <- col (D.refine (\case "refuse" -> Right RefuseDeadLetters; "remove" -> Right RemoveDeadLetters; _ -> Left "invalid dead-letter policy") D.text)+  cause <- col (D.refine (\case "refuse" -> Right RefuseCausationDependents; "allow" -> Right AllowDanglingCausation; _ -> Left "invalid causation policy") D.text)+  selected <- col D.int8+  home <- col D.int8+  global <- col D.int8+  links <- col D.int8+  deadLetters <- col D.int8+  dependents <- col D.int8+  lo <- GlobalPosition <$> col D.int8+  hi <- GlobalPosition <$> col D.int8+  streams <- col (D.refine parseHeads D.jsonb)+  applied <- col D.timestamptz+  actor <- col D.text+  pure (CompactionRecord cid (CompactionReport md identity operation dead cause selected home global links deadLetters dependents lo hi streams rd) applied actor)+  where+    digestValue = D.refine (maybe (Left "invalid SHA-256 byte length") Right . mkCompactionDigest) D.bytea+    parseHeads value = case Aeson.fromJSON value of+      Aeson.Error err -> Left (T.pack err)+      Aeson.Success heads -> Right heads++col :: D.Value a -> D.Row a+col = D.column . D.nonNullable++lockCompactionStreamsStmt :: Statement (Vector Text) (Vector (Text, Int64, Int64, Maybe UTCTime))+lockCompactionStreamsStmt =+  preparable+    "SELECT stream_name,stream_id,stream_version,deleted_at FROM streams WHERE stream_name = ANY($1::text[]) ORDER BY stream_id FOR UPDATE"+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.text))))+    (D.rowVector ((,,,) <$> col D.text <*> col D.int8 <*> col D.int8 <*> D.column (D.nullable D.timestamptz)))++deleteJunctionRowsStmt :: Statement (Vector UUID, Vector Int64) (Vector (Int64, Int64))+deleteJunctionRowsStmt =+  preparable+    "DELETE FROM stream_events se USING unnest($1::uuid[], $2::bigint[]) AS k(event_id,stream_id) WHERE se.event_id=k.event_id AND se.stream_id=k.stream_id RETURNING se.stream_id,se.original_stream_id"+    (contrazip2 (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid)))) (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.int8)))))+    (D.rowVector ((,) <$> col D.int8 <*> col D.int8))++deleteSelectedDeadLettersStmt :: Statement (Vector UUID) Int64+deleteSelectedDeadLettersStmt =+  preparable+    "WITH removed AS (DELETE FROM dead_letters WHERE event_id=ANY($1::uuid[]) RETURNING 1) SELECT count(*) FROM removed"+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    (D.singleRow (col D.int8))++deleteSelectedEventsStmt :: Statement (Vector UUID) Int64+deleteSelectedEventsStmt =+  preparable+    "WITH removed AS (DELETE FROM events WHERE event_id=ANY($1::uuid[]) RETURNING 1) SELECT count(*) FROM removed"+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    (D.singleRow (col D.int8))++compactionLedgerStmt :: Statement Int32 (Vector CompactionRecord)+compactionLedgerStmt =+  preparable+    ("SELECT " <> ledgerColumns <> " FROM event_compactions ORDER BY applied_at DESC,compaction_id LIMIT $1")+    (E.param (E.nonNullable E.int4))+    (D.rowVector compactionRecordRow)++data CompactionRecordInsert = CompactionRecordInsert+  { manifestDigest :: !CompactionDigest,+    reportDigest :: !CompactionDigest,+    storeId :: !StoreIdentity,+    operation :: !CompactionOperation,+    deadLetterPolicy :: !DeadLetterPolicy,+    causationPolicy :: !CausationPolicy,+    selectedEvents :: !Int64,+    homeMemberships :: !Int64,+    globalMemberships :: !Int64,+    linkMemberships :: !Int64,+    deadLettersRemoved :: !Int64,+    causationDependents :: !Int64,+    lowestGlobalPosition :: !GlobalPosition,+    highestGlobalPosition :: !GlobalPosition,+    affectedStreams :: !(Vector StreamHeadWitness)+  }+  deriving stock (Eq, Show, Generic)++insertCompactionRecordStmt :: Statement CompactionRecordInsert (UUID, UTCTime, Text)+insertCompactionRecordStmt =+  preparable+    """+    INSERT INTO event_compactions (manifest_digest,report_digest,store_id,operation,+      dead_letter_policy,causation_policy,selected_events,home_memberships,+      global_memberships,link_memberships,dead_letters_removed,causation_dependents,+      lowest_global_position,highest_global_position,affected_streams)+    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)+    RETURNING compaction_id,applied_at,applied_by+    """+    ( ((\r -> compactionDigestBytes r.manifestDigest) >$< E.param (E.nonNullable E.bytea))+        <> ((\r -> compactionDigestBytes r.reportDigest) >$< E.param (E.nonNullable E.bytea))+        <> ((\r -> let StoreIdentity uid = r.storeId in uid) >$< E.param (E.nonNullable E.uuid))+        <> ((\r -> compactionOperationText r.operation) >$< E.param (E.nonNullable E.text))+        <> ((\r -> if r.deadLetterPolicy == RefuseDeadLetters then "refuse" else "remove") >$< E.param (E.nonNullable E.text))+        <> ((\r -> if r.causationPolicy == RefuseCausationDependents then "refuse" else "allow") >$< E.param (E.nonNullable E.text))+        <> ((\r -> r.selectedEvents) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> r.homeMemberships) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> r.globalMemberships) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> r.linkMemberships) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> r.deadLettersRemoved) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> r.causationDependents) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> let GlobalPosition n = r.lowestGlobalPosition in n) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> let GlobalPosition n = r.highestGlobalPosition in n) >$< E.param (E.nonNullable E.int8))+        <> ((\r -> Aeson.toJSON r.affectedStreams) >$< E.param (E.nonNullable E.jsonb))+    )+    (D.singleRow ((,,) <$> col D.uuid <*> col D.timestamptz <*> col D.text))

@@ -0,0 +1,672 @@+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE NoFieldSelectors #-}++-- | Digest-sealed, explicitly witnessed selective compaction inputs and outcomes.+-- The consumer proves redundancy and archives envelopes. These types describe+-- physical deletion of exact events while preserving stream heads and retained+-- versions; logical truncate-before and whole-stream hard delete are separate+-- lifecycle operations. File-space reclamation is PostgreSQL maintenance.+-- Canonical bytes are a versioned cross-language contract, independent of JSON.+module Kiroku.Store.Compaction.Types+  ( CompactionOperation,+    CompactionDigest,+    CompactionId (..),+    LinkWitness (..),+    StreamHeadWitness (..),+    CompactionSelection (..),+    DeadLetterPolicy (..),+    CausationPolicy (..),+    CompactionManifestInput (..),+    CompactionManifestError (..),+    CompactionRefusal (..),+    WitnessMismatch (..),+    CompactionReport (..),+    CompactionRecord (..),+    CompactionPreview (..),+    CompactionApplyResult (..),+    CompactionLedgerLimit,+    CompactionLedgerQuery (..),+    CompactionManifest,+    CompactionLedgerError (..),+    mkCompactionManifest,+    compactionManifestDigest,+    canonicalCompactionManifestBytes,+    maxCompactionSelections,+    mkCompactionOperation,+    compactionOperationText,+    mkCompactionDigest,+    compactionDigestBytes,+    compactionDigestHex,+    parseCompactionDigestHex,+    mkCompactionLedgerLimit,+    compactionLedgerLimitValue,+    sealCompactionReport,+    compactionReportDigest,+    canonicalCompactionReportBytes,+    manifestStoreIdentity,+    manifestOperation,+    manifestDeadLetterPolicy,+    manifestCausationPolicy,+    manifestStreamHeads,+    manifestSelections,+    manifestStreamNames,+  )+where++import Control.Monad (unless, when)+import Crypto.Hash (Digest, SHA256, hash)+import Data.Aeson (FromJSON (..), ToJSON (..), Value (String), object, withObject, withText, (.:), (.:?), (.=))+import Data.Aeson.Types (Parser)+import Data.ByteArray (convert)+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.ByteString.Builder+import Data.ByteString.Lazy qualified as LBS+import Data.Char (digitToInt, intToDigit)+import Data.Int (Int32, Int64)+import Data.List (sortOn)+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.Time (UTCTime)+import Data.UUID (UUID)+import Data.UUID qualified as UUID+import Data.Vector (Vector)+import Data.Vector qualified as V+import GHC.Generics (Generic)+import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict (..))+import Kiroku.Store.Types++newtype CompactionOperation = CompactionOperation Text -- validated 1..512 UTF-8 bytes+  deriving stock (Eq, Ord, Show, Generic)++newtype CompactionDigest = CompactionDigest ByteString -- exactly 32 bytes (SHA-256)+  deriving stock (Eq, Ord, Show, Generic)++newtype CompactionId = CompactionId UUID+  deriving stock (Eq, Ord, Show, Generic)++data LinkWitness = LinkWitness {stream :: !StreamName, streamVersion :: !StreamVersion}+  deriving stock (Eq, Show, Generic)++data StreamHeadWitness = StreamHeadWitness {stream :: !StreamName, headVersion :: !StreamVersion}+  deriving stock (Eq, Show, Generic)++data CompactionSelection = CompactionSelection+  { eventId :: !EventId,+    originStream :: !StreamName,+    originVersion :: !StreamVersion,+    globalPosition :: !GlobalPosition,+    acknowledgedLinks :: !(Vector LinkWitness) -- sorted by (stream, streamVersion), unique+  }+  deriving stock (Eq, Show, Generic)++data DeadLetterPolicy = RefuseDeadLetters | RemoveDeadLetters+  deriving stock (Eq, Ord, Show, Enum, Bounded, Generic)++data CausationPolicy = RefuseCausationDependents | AllowDanglingCausation+  deriving stock (Eq, Ord, Show, Enum, Bounded, Generic)++data CompactionManifestInput = CompactionManifestInput+  { storeIdentity :: !StoreIdentity,+    operation :: !CompactionOperation,+    deadLetterPolicy :: !DeadLetterPolicy,+    causationPolicy :: !CausationPolicy,+    streamHeads :: ![StreamHeadWitness],+    selections :: ![CompactionSelection],+    expectedDigest :: !(Maybe CompactionDigest)+  }+  deriving stock (Eq, Show, Generic)++data CompactionManifestError+  = CompactionManifestEmpty+  | CompactionManifestTooLarge !Int+  | CompactionManifestDuplicateEvent !EventId+  | CompactionManifestReservedStream !StreamName+  | CompactionManifestInvalidStreamName !StreamName !Int+  | CompactionManifestNonPositiveVersion !EventId+  | CompactionManifestNonPositivePosition !EventId+  | CompactionManifestDuplicatePosition !GlobalPosition+  | CompactionManifestDuplicateOriginVersion !StreamName !StreamVersion+  | CompactionManifestDuplicateLink !EventId !LinkWitness+  | CompactionManifestLinkIntoOrigin !EventId+  | CompactionManifestMissingStreamHead !StreamName+  | CompactionManifestUnusedStreamHead !StreamName+  | CompactionManifestDuplicateStreamHead !StreamName+  | CompactionManifestOperationEmpty+  | CompactionManifestOperationTooLong !Int+  | CompactionManifestDigestMismatch {expected :: !CompactionDigest, computed :: !CompactionDigest}+  deriving stock (Eq, Show, Generic)++data CompactionRefusal+  = CompactionStoreIdentityMismatch {expectedIdentity :: !StoreIdentity, actualIdentity :: !StoreIdentity}+  | CompactionHistoryRetentionActive !HistoryRetentionConflict+  | CompactionStreamMissing !StreamName+  | CompactionStreamSoftDeleted !StreamName+  | CompactionStreamHeadDrift {stream :: !StreamName, expectedHead :: !StreamVersion, actualHead :: !StreamVersion}+  | CompactionSelectedEventMissing !EventId+  | CompactionWitnessMismatch !EventId !WitnessMismatch+  | CompactionUnexpectedLink !EventId !LinkWitness+  | CompactionAcknowledgedLinkMissing !EventId !LinkWitness+  | CompactionDeadLettersPresent !EventId !Int64+  | CompactionCausationDependentsPresent !EventId !Int64+  | CompactionLedgerConflict {compactionId :: !CompactionId, survivingEvents :: !Int64}+  deriving stock (Eq, Show, Generic)++data WitnessMismatch = WitnessMismatch+  { actualOriginStream :: !StreamName,+    actualOriginVersion :: !StreamVersion,+    actualGlobalPosition :: !(Maybe GlobalPosition) -- Nothing when the $all row is gone+  }+  deriving stock (Eq, Show, Generic)++data CompactionReport = CompactionReport+  { manifestDigest :: !CompactionDigest,+    storeIdentity :: !StoreIdentity,+    operation :: !CompactionOperation,+    deadLetterPolicy :: !DeadLetterPolicy,+    causationPolicy :: !CausationPolicy,+    selectedEvents :: !Int64,+    homeMemberships :: !Int64,+    globalMemberships :: !Int64,+    linkMemberships :: !Int64,+    deadLettersRemoved :: !Int64,+    causationDependents :: !Int64,+    lowestGlobalPosition :: !GlobalPosition,+    highestGlobalPosition :: !GlobalPosition,+    affectedStreams :: !(Vector StreamHeadWitness), -- sorted by stream name+    reportDigest :: !CompactionDigest+  }+  deriving stock (Eq, Show, Generic)++data CompactionRecord = CompactionRecord+  { compactionId :: !CompactionId,+    report :: !CompactionReport,+    appliedAt :: !UTCTime,+    appliedBy :: !Text+  }+  deriving stock (Eq, Show, Generic)++data CompactionPreview+  = CompactionPreviewReady !CompactionReport+  | CompactionPreviewAlreadyApplied !CompactionRecord+  deriving stock (Eq, Show, Generic)++data CompactionApplyResult+  = CompactionAppliedNow !CompactionRecord+  | CompactionAlreadyApplied !CompactionRecord+  deriving stock (Eq, Show, Generic)++newtype CompactionLedgerLimit = CompactionLedgerLimit Int32 -- validated 1..1000+  deriving stock (Eq, Ord, Show, Generic)++data CompactionLedgerQuery = CompactionLedgerQuery {limit :: !CompactionLedgerLimit}+  deriving stock (Eq, Show, Generic)++data CompactionLedgerError = CompactionLedgerLimitOutOfRange !Int32+  deriving stock (Eq, Show, Generic)++data CompactionManifest = CompactionManifest+  { identityValue :: !StoreIdentity,+    operationValue :: !CompactionOperation,+    deadPolicyValue :: !DeadLetterPolicy,+    causationPolicyValue :: !CausationPolicy,+    headValues :: !(Vector StreamHeadWitness),+    selectionValues :: !(Vector CompactionSelection)+  }+  deriving stock (Eq, Show, Generic)++manifestStoreIdentity :: CompactionManifest -> StoreIdentity+manifestStoreIdentity m = m.identityValue++manifestOperation :: CompactionManifest -> CompactionOperation+manifestOperation m = m.operationValue++manifestDeadLetterPolicy :: CompactionManifest -> DeadLetterPolicy+manifestDeadLetterPolicy m = m.deadPolicyValue++manifestCausationPolicy :: CompactionManifest -> CausationPolicy+manifestCausationPolicy m = m.causationPolicyValue++manifestStreamHeads :: CompactionManifest -> Vector StreamHeadWitness+manifestStreamHeads m = m.headValues++manifestSelections :: CompactionManifest -> Vector CompactionSelection+manifestSelections m = m.selectionValues++manifestStreamNames :: CompactionManifest -> Vector Text+manifestStreamNames m = V.map (\h -> let StreamName name = h.stream in name) m.headValues++maxCompactionSelections :: Int+maxCompactionSelections = 100000++mkCompactionOperation :: Text -> Either CompactionManifestError CompactionOperation+mkCompactionOperation value+  | bytes == 0 = Left CompactionManifestOperationEmpty+  | bytes > 512 = Left (CompactionManifestOperationTooLong bytes)+  | otherwise = Right (CompactionOperation value)+  where+    bytes = BS.length (TE.encodeUtf8 value)++compactionOperationText :: CompactionOperation -> Text+compactionOperationText (CompactionOperation t) = t++mkCompactionDigest :: ByteString -> Maybe CompactionDigest+mkCompactionDigest b+  | BS.length b == 32 = Just (CompactionDigest b)+  | otherwise = Nothing++compactionDigestBytes :: CompactionDigest -> ByteString+compactionDigestBytes (CompactionDigest b) = b++compactionDigestHex :: CompactionDigest -> Text+compactionDigestHex (CompactionDigest b) = T.pack (concatMap hex (BS.unpack b))+  where+    hex n = [intToDigit (fromIntegral n `div` 16), intToDigit (fromIntegral n `mod` 16)]++parseCompactionDigestHex :: Text -> Maybe CompactionDigest+parseCompactionDigestHex t+  | T.length t /= 64 || not (T.all (\c -> c `elem` ("0123456789abcdefABCDEF" :: String)) t) = Nothing+  | otherwise = mkCompactionDigest (BS.pack (pairs (T.unpack t)))+  where+    pairs (a : b : rest) = fromIntegral (16 * digitToInt a + digitToInt b) : pairs rest+    pairs _ = []++mkCompactionLedgerLimit :: Int32 -> Either CompactionLedgerError CompactionLedgerLimit+mkCompactionLedgerLimit n+  | n >= 1 && n <= 1000 = Right (CompactionLedgerLimit n)+  | otherwise = Left (CompactionLedgerLimitOutOfRange n)++compactionLedgerLimitValue :: CompactionLedgerLimit -> Int32+compactionLedgerLimitValue (CompactionLedgerLimit n) = n++-- Kept below Store.Error in the import graph: Error consumes the digest type.+-- These are the same reserved-name and UTF-8 length rules as ordinary writes.+validateName :: StreamName -> Either CompactionManifestError ()+validateName name@(StreamName text)+  | text == "$all" = Left (CompactionManifestReservedStream name)+  | bytes > 512 = Left (CompactionManifestInvalidStreamName name bytes)+  | otherwise = Right ()+  where+    bytes = BS.length (TE.encodeUtf8 text)++uniqueBy :: (Ord b) => (a -> b) -> (a -> CompactionManifestError) -> [a] -> Either CompactionManifestError ()+uniqueBy key err = go Set.empty+  where+    go _ [] = Right ()+    go seen (x : xs)+      | Set.member (key x) seen = Left (err x)+      | otherwise = go (Set.insert (key x) seen) xs++mkCompactionManifest :: CompactionManifestInput -> Either CompactionManifestError CompactionManifest+mkCompactionManifest input = do+  let sels = input.selections+      heads = input.streamHeads+      CompactionOperation op = input.operation+  _ <- mkCompactionOperation op+  when (null sels) (Left CompactionManifestEmpty)+  when (length sels > maxCompactionSelections) (Left (CompactionManifestTooLarge (length sels)))+  mapM_ validateSelection sels+  uniqueBy (\s -> s.eventId) (\s -> CompactionManifestDuplicateEvent s.eventId) sels+  uniqueBy (\s -> s.globalPosition) (\s -> CompactionManifestDuplicatePosition s.globalPosition) sels+  uniqueBy (\s -> (s.originStream, s.originVersion)) (\s -> CompactionManifestDuplicateOriginVersion s.originStream s.originVersion) sels+  mapM_ (\h -> validateName h.stream) heads+  uniqueBy (\h -> h.stream) (\h -> CompactionManifestDuplicateStreamHead h.stream) heads+  let touched = Set.fromList (concatMap (\s -> s.originStream : map (\l -> l.stream) (V.toList s.acknowledgedLinks)) sels)+      declared = Set.fromList (map (\h -> h.stream) heads)+  case Set.lookupMin (touched Set.\\ declared) of+    Just name -> Left (CompactionManifestMissingStreamHead name)+    Nothing -> pure ()+  case Set.lookupMin (declared Set.\\ touched) of+    Just name -> Left (CompactionManifestUnusedStreamHead name)+    Nothing -> pure ()+  let canonicalSelection s = s {acknowledgedLinks = V.fromList (sortOn linkKey (V.toList s.acknowledgedLinks))}+      m =+        CompactionManifest+          input.storeIdentity+          input.operation+          input.deadLetterPolicy+          input.causationPolicy+          (V.fromList (sortOn (\h -> nameBytes h.stream) heads))+          (V.fromList (map canonicalSelection (sortOn (\s -> s.globalPosition) sels)))+      computed = compactionManifestDigest m+  case input.expectedDigest of+    Just expected | expected /= computed -> Left (CompactionManifestDigestMismatch expected computed)+    _ -> Right m+  where+    nameBytes (StreamName t) = TE.encodeUtf8 t+    linkKey l = (nameBytes l.stream, l.streamVersion)+    validateSelection s = do+      validateName s.originStream+      let StreamVersion ver = s.originVersion+          GlobalPosition pos = s.globalPosition+      when (ver < 1) (Left (CompactionManifestNonPositiveVersion s.eventId))+      when (pos < 1) (Left (CompactionManifestNonPositivePosition s.eventId))+      mapM_ (validateLink s) s.acknowledgedLinks+      uniqueBy linkKey (CompactionManifestDuplicateLink s.eventId) (V.toList s.acknowledgedLinks)+    validateLink s l = do+      validateName l.stream+      let StreamVersion ver = l.streamVersion+      when (ver < 1) (Left (CompactionManifestNonPositiveVersion s.eventId))+      when (l.stream == s.originStream) (Left (CompactionManifestLinkIntoOrigin s.eventId))++sha256 :: ByteString -> CompactionDigest+sha256 bytes = CompactionDigest (convert (hash bytes :: Digest SHA256))++lengthPrefixed :: Text -> Builder+lengthPrefixed t = let b = TE.encodeUtf8 t in word32BE (fromIntegral (BS.length b)) <> byteString b++uuidBytes :: UUID -> Builder+uuidBytes = lazyByteString . UUID.toByteString++counted :: Vector a -> (a -> Builder) -> Builder+counted values encode = word32BE (fromIntegral (V.length values)) <> foldMap encode values++headBytes :: StreamHeadWitness -> Builder+headBytes (StreamHeadWitness (StreamName name) (StreamVersion ver)) = lengthPrefixed name <> int64BE ver++linkBytes :: LinkWitness -> Builder+linkBytes (LinkWitness (StreamName name) (StreamVersion ver)) = lengthPrefixed name <> int64BE ver++policyBytes :: DeadLetterPolicy -> CausationPolicy -> Builder+policyBytes dead cause = word8 (if dead == RefuseDeadLetters then 0 else 1) <> word8 (if cause == RefuseCausationDependents then 0 else 1)++-- | Version-1 digest protocol, independent of JSON: ASCII format tag plus LF,+-- raw 16-byte store UUID, length-prefixed UTF-8 operation, two policy bytes+-- (0 refuses, 1 allows/removes), counted heads, then counted selections. Text+-- lengths and vector counts are unsigned big-endian Word32; versions and+-- positions are signed big-endian Int64. Heads sort by UTF-8 name; selections+-- by global position; links by UTF-8 name then version. A head/link is name+-- then version. A selection is raw UUID, origin name, origin version, global+-- position and counted links. See @docs/user/compaction.md@ for the test vector.+canonicalCompactionManifestBytes :: CompactionManifest -> ByteString+canonicalCompactionManifestBytes m =+  LBS.toStrict . toLazyByteString $+    byteString "kiroku-compaction-manifest/1\n"+      <> uuidBytes uid+      <> lengthPrefixed (compactionOperationText m.operationValue)+      <> policyBytes m.deadPolicyValue m.causationPolicyValue+      <> counted m.headValues headBytes+      <> counted m.selectionValues selectionBytes+  where+    StoreIdentity uid = m.identityValue+    selectionBytes s =+      let EventId eid = s.eventId+          StreamName name = s.originStream+          StreamVersion ver = s.originVersion+          GlobalPosition pos = s.globalPosition+       in uuidBytes eid <> lengthPrefixed name <> int64BE ver <> int64BE pos <> counted s.acknowledgedLinks linkBytes++compactionManifestDigest :: CompactionManifest -> CompactionDigest+compactionManifestDigest = sha256 . canonicalCompactionManifestBytes++-- | Version-1 report protocol: format tag plus LF, raw manifest digest, store+-- UUID, operation and policies, then eight Int64 values: selected, home, global,+-- link, removed dead-letter and causation counts, lowest and highest position.+-- Counted affected heads follow, using the manifest encoding. The report's own+-- digest is excluded. Stored ledger digests remain verbatim audit evidence.+canonicalCompactionReportBytes :: CompactionReport -> ByteString+canonicalCompactionReportBytes r =+  LBS.toStrict . toLazyByteString $+    byteString "kiroku-compaction-report/1\n"+      <> byteString (compactionDigestBytes r.manifestDigest)+      <> uuidBytes uid+      <> lengthPrefixed (compactionOperationText r.operation)+      <> policyBytes r.deadLetterPolicy r.causationPolicy+      <> foldMap+        int64BE+        [ r.selectedEvents,+          r.homeMemberships,+          r.globalMemberships,+          r.linkMemberships,+          r.deadLettersRemoved,+          r.causationDependents,+          lo,+          hi+        ]+      <> counted r.affectedStreams headBytes+  where+    StoreIdentity uid = r.storeIdentity+    GlobalPosition lo = r.lowestGlobalPosition+    GlobalPosition hi = r.highestGlobalPosition++compactionReportDigest :: CompactionReport -> CompactionDigest+compactionReportDigest = sha256 . canonicalCompactionReportBytes++sealCompactionReport :: CompactionReport -> CompactionReport+sealCompactionReport r = r {reportDigest = compactionReportDigest r}++instance ToJSON CompactionId where+  toJSON (CompactionId uid) = toJSON uid++instance FromJSON CompactionId where+  parseJSON value = CompactionId <$> parseJSON value++instance ToJSON CompactionDigest where+  toJSON = toJSON . compactionDigestHex++instance FromJSON CompactionDigest where+  parseJSON = withText "CompactionDigest" $ \t -> maybe (fail "expected a 64-character SHA-256 hex digest") pure (parseCompactionDigestHex t)++instance ToJSON CompactionOperation where+  toJSON = toJSON . compactionOperationText++instance FromJSON CompactionOperation where+  parseJSON = withText "CompactionOperation" $ either (fail . show) pure . mkCompactionOperation++instance ToJSON DeadLetterPolicy where+  toJSON p = String (if p == RefuseDeadLetters then "refuse" else "remove")++instance FromJSON DeadLetterPolicy where+  parseJSON = withText "DeadLetterPolicy" $ \case+    "refuse" -> pure RefuseDeadLetters+    "remove" -> pure RemoveDeadLetters+    _ -> fail "dead_letter_policy must be refuse or remove"++instance ToJSON CausationPolicy where+  toJSON p = String (if p == RefuseCausationDependents then "refuse" else "allow")++instance FromJSON CausationPolicy where+  parseJSON = withText "CausationPolicy" $ \case+    "refuse" -> pure RefuseCausationDependents+    "allow" -> pure AllowDanglingCausation+    _ -> fail "causation_policy must be refuse or allow"++streamText :: StreamName -> Text+streamText (StreamName name) = name++versionInt :: StreamVersion -> Int64+versionInt (StreamVersion ver) = ver++positionInt :: GlobalPosition -> Int64+positionInt (GlobalPosition pos) = pos++eventUuid :: EventId -> UUID+eventUuid (EventId uid) = uid++instance ToJSON LinkWitness where+  toJSON r =+    object+      [ "stream" .= (streamText r.stream),+        "stream_version" .= (versionInt r.streamVersion)+      ]++instance FromJSON LinkWitness where+  parseJSON = withObject "LinkWitness" $ \o ->+    LinkWitness+      <$> (StreamName <$> o .: "stream")+      <*> (StreamVersion <$> o .: "stream_version")++instance ToJSON StreamHeadWitness where+  toJSON r =+    object+      [ "stream" .= (streamText r.stream),+        "head_version" .= (versionInt r.headVersion)+      ]++instance FromJSON StreamHeadWitness where+  parseJSON = withObject "StreamHeadWitness" $ \o ->+    StreamHeadWitness+      <$> (StreamName <$> o .: "stream")+      <*> (StreamVersion <$> o .: "head_version")++instance ToJSON CompactionSelection where+  toJSON r =+    object+      [ "event_id" .= (eventUuid r.eventId),+        "origin_stream" .= (streamText r.originStream),+        "origin_version" .= (versionInt r.originVersion),+        "global_position" .= (positionInt r.globalPosition),+        "acknowledged_links" .= (r.acknowledgedLinks)+      ]++instance FromJSON CompactionSelection where+  parseJSON = withObject "CompactionSelection" $ \o ->+    CompactionSelection+      <$> (EventId <$> o .: "event_id")+      <*> (StreamName <$> o .: "origin_stream")+      <*> (StreamVersion <$> o .: "origin_version")+      <*> (GlobalPosition <$> o .: "global_position")+      <*> o .: "acknowledged_links"++instance ToJSON WitnessMismatch where+  toJSON r =+    object+      [ "actual_origin_stream" .= (streamText r.actualOriginStream),+        "actual_origin_version" .= (versionInt r.actualOriginVersion),+        "actual_global_position" .= (fmap positionInt r.actualGlobalPosition)+      ]++instance FromJSON WitnessMismatch where+  parseJSON = withObject "WitnessMismatch" $ \o ->+    WitnessMismatch+      <$> (StreamName <$> o .: "actual_origin_stream")+      <*> (StreamVersion <$> o .: "actual_origin_version")+      <*> (fmap GlobalPosition <$> o .: "actual_global_position")++instance ToJSON CompactionReport where+  toJSON r =+    object+      [ "manifest_digest" .= (r.manifestDigest),+        "store_identity" .= (r.storeIdentity),+        "operation" .= (r.operation),+        "dead_letter_policy" .= (r.deadLetterPolicy),+        "causation_policy" .= (r.causationPolicy),+        "selected_events" .= (r.selectedEvents),+        "home_memberships" .= (r.homeMemberships),+        "global_memberships" .= (r.globalMemberships),+        "link_memberships" .= (r.linkMemberships),+        "dead_letters_removed" .= (r.deadLettersRemoved),+        "causation_dependents" .= (r.causationDependents),+        "lowest_global_position" .= (positionInt r.lowestGlobalPosition),+        "highest_global_position" .= (positionInt r.highestGlobalPosition),+        "affected_streams" .= (r.affectedStreams),+        "report_digest" .= (r.reportDigest)+      ]++instance FromJSON CompactionReport where+  parseJSON = withObject "CompactionReport" $ \o -> do+    r <-+      CompactionReport+        <$> o .: "manifest_digest"+        <*> o .: "store_identity"+        <*> o .: "operation"+        <*> o .: "dead_letter_policy"+        <*> o .: "causation_policy"+        <*> o .: "selected_events"+        <*> o .: "home_memberships"+        <*> o .: "global_memberships"+        <*> o .: "link_memberships"+        <*> o .: "dead_letters_removed"+        <*> o .: "causation_dependents"+        <*> (GlobalPosition <$> o .: "lowest_global_position")+        <*> (GlobalPosition <$> o .: "highest_global_position")+        <*> o .: "affected_streams"+        <*> o .: "report_digest"+    unless (compactionReportDigest r == r.reportDigest) (fail "compaction report digest mismatch")+    pure r++instance ToJSON CompactionRecord where+  toJSON r =+    object+      [ "compaction_id" .= (r.compactionId),+        "report" .= (r.report),+        "applied_at" .= (r.appliedAt),+        "applied_by" .= (r.appliedBy)+      ]++instance FromJSON CompactionRecord where+  parseJSON = withObject "CompactionRecord" $ \o ->+    CompactionRecord+      <$> o .: "compaction_id"+      <*> o .: "report"+      <*> o .: "applied_at"+      <*> o .: "applied_by"++instance ToJSON CompactionManifest where+  toJSON m =+    object+      [ "format" .= ("kiroku-compaction-manifest/1" :: Text),+        "store_identity" .= manifestStoreIdentity m,+        "operation" .= manifestOperation m,+        "dead_letter_policy" .= manifestDeadLetterPolicy m,+        "causation_policy" .= manifestCausationPolicy m,+        "stream_heads" .= manifestStreamHeads m,+        "selections" .= manifestSelections m,+        "digest" .= compactionManifestDigest m+      ]++instance FromJSON CompactionManifest where+  parseJSON = withObject "CompactionManifest" $ \o -> do+    format <- o .: "format"+    unless (format == ("kiroku-compaction-manifest/1" :: Text)) (fail "unsupported compaction manifest format")+    digest <- o .:? "digest" >>= maybe (fail "compaction manifest requires a digest") pure+    input <-+      CompactionManifestInput+        <$> o .: "store_identity"+        <*> o .: "operation"+        <*> o .: "dead_letter_policy"+        <*> o .: "causation_policy"+        <*> o .: "stream_heads"+        <*> o .: "selections"+        <*> pure (Just digest)+    either (fail . show) pure (mkCompactionManifest input)++instance ToJSON CompactionRefusal where+  toJSON refusal = object $ case refusal of+    CompactionStoreIdentityMismatch expected actual -> tag "store_identity_mismatch" <> ["expected" .= expected, "actual" .= actual]+    CompactionHistoryRetentionActive (HistoryRetentionConflict count expiry) -> tag "history_retention_active" <> ["active_lease_count" .= count, "earliest_expiry" .= expiry]+    CompactionStreamMissing stream -> tag "stream_missing" <> ["stream" .= streamText stream]+    CompactionStreamSoftDeleted stream -> tag "stream_soft_deleted" <> ["stream" .= streamText stream]+    CompactionStreamHeadDrift stream expected actual -> tag "stream_head_drift" <> ["stream" .= streamText stream, "expected" .= versionInt expected, "actual" .= versionInt actual]+    CompactionSelectedEventMissing eid -> tag "selected_event_missing" <> ["event_id" .= eventUuid eid]+    CompactionWitnessMismatch eid witness -> tag "witness_mismatch" <> ["event_id" .= eventUuid eid, "witness" .= witness]+    CompactionUnexpectedLink eid link -> tag "unexpected_link" <> ["event_id" .= eventUuid eid, "link" .= link]+    CompactionAcknowledgedLinkMissing eid link -> tag "acknowledged_link_missing" <> ["event_id" .= eventUuid eid, "link" .= link]+    CompactionDeadLettersPresent eid count -> tag "dead_letters_present" <> ["event_id" .= eventUuid eid, "count" .= count]+    CompactionCausationDependentsPresent eid count -> tag "causation_dependents_present" <> ["event_id" .= eventUuid eid, "count" .= count]+    CompactionLedgerConflict cid count -> tag "ledger_conflict" <> ["compaction_id" .= cid, "surviving_events" .= count]+    where+      tag name = ["refusal" .= (name :: Text)]++instance FromJSON CompactionRefusal where+  parseJSON = withObject "CompactionRefusal" $ \o -> do+    kind <- o .: "refusal" :: Parser Text+    let stream = StreamName <$> o .: "stream"+        eid = EventId <$> o .: "event_id"+    case kind of+      "store_identity_mismatch" -> CompactionStoreIdentityMismatch <$> o .: "expected" <*> o .: "actual"+      "history_retention_active" -> CompactionHistoryRetentionActive <$> (HistoryRetentionConflict <$> o .: "active_lease_count" <*> o .: "earliest_expiry")+      "stream_missing" -> CompactionStreamMissing <$> stream+      "stream_soft_deleted" -> CompactionStreamSoftDeleted <$> stream+      "stream_head_drift" -> CompactionStreamHeadDrift <$> stream <*> (StreamVersion <$> o .: "expected") <*> (StreamVersion <$> o .: "actual")+      "selected_event_missing" -> CompactionSelectedEventMissing <$> eid+      "witness_mismatch" -> CompactionWitnessMismatch <$> eid <*> o .: "witness"+      "unexpected_link" -> CompactionUnexpectedLink <$> eid <*> o .: "link"+      "acknowledged_link_missing" -> CompactionAcknowledgedLinkMissing <$> eid <*> o .: "link"+      "dead_letters_present" -> CompactionDeadLettersPresent <$> eid <*> o .: "count"+      "causation_dependents_present" -> CompactionCausationDependentsPresent <$> eid <*> o .: "count"+      "ledger_conflict" -> CompactionLedgerConflict <$> o .: "compaction_id" <*> o .: "surviving_events"+      _ -> fail "unknown compaction refusal"

@@ -17,6 +17,9 @@     prepareEvents,     buildAppendParams,     appendDispatchTx,+    assembleEventReferences,+    chunkUuids,+    maxEventReferenceBatch,   ) where @@ -25,8 +28,10 @@ import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value) import Data.Generics.Labels ()-import Data.Int (Int32)+import Data.Int (Int32, Int64) import Data.List (find)+import Data.List.NonEmpty (NonEmpty)+import Data.List.NonEmpty qualified as NonEmpty import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Maybe (isNothing)@@ -38,7 +43,7 @@ import Data.Vector qualified as V import Effectful (Dispatch (..), DispatchOf, Eff, Effect, IOE, runEff, (:>)) import Effectful.Dispatch.Dynamic (interpret_)-import Effectful.Error.Static (Error, runErrorNoCallStack, throwError)+import Effectful.Error.Static (Error, catchError, runErrorNoCallStack, throwError) import GHC.Generics (Generic) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E@@ -49,6 +54,8 @@ import Hasql.Statement (Statement, preparable) import Hasql.Transaction qualified as Tx import Hasql.Transaction.Sessions qualified as TxSessions+import Kiroku.Store.Compaction.Internal qualified as Compaction+import Kiroku.Store.Compaction.Types import Kiroku.Store.Connection (KirokuStore (..)) import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore) import Kiroku.Store.Error (StoreError (..), attributeMultiStreamError, emptyResultError, isTransientSerializationError, mapGenericUsageError, mapLinkUsageError, mapTransactionUsageError, mapUsageError, validateStreamName)@@ -89,6 +96,12 @@   --   --     Surfaced as 'Kiroku.Store.Read.visibleGlobalHeadPosition'.   GetVisibleGlobalHeadPosition :: Store m GlobalPosition+  -- | Read the immutable store identity installed by migration 0017.+  GetStoreIdentity :: Store m StoreIdentity+  -- | Read-only manifest preview on one repeatable-read snapshot, without row locks.+  PreviewCompaction :: CompactionManifest -> Store m (Either (NonEmpty CompactionRefusal) CompactionPreview)+  ApplyCompaction :: CompactionManifest -> Store m (Either (NonEmpty CompactionRefusal) CompactionApplyResult)+  GetCompactionLedger :: CompactionLedgerQuery -> Store m (Vector CompactionRecord)   GetStream :: StreamName -> Store m (Maybe StreamInfo)   -- | One-statement metadata and newest surviving originated head observation.   --     Outer 'Nothing' means absent; inner 'Nothing' means no originated events,@@ -120,6 +133,8 @@   --     the distinct ids from a batch and resolve them once, rather than per event.   --     Surfaced as 'Kiroku.Store.Read.lookupStreamNames' (and the singular   --     'Kiroku.Store.Read.lookupStreamName').+  -- | Inventory of memberships and references outside the requested event set.+  LookupEventReferences :: [EventId] -> Store m (Map EventId EventReferenceInventory)   LookupStreamNames :: [StreamId] -> Store m (Map StreamId StreamName)   -- | Byte-ordered names; optional exact category, literal prefix and exclusive cursor.   --     Includes soft-deleted streams, excludes only the reserved stream row.@@ -262,6 +277,40 @@       usePool (store ^. #pool) $         Session.statement (cursor, limit) SQL.readAllBackwardStmt     decodeReadEvents (store ^. #storeSettings) evs+  ApplyCompaction manifest -> do+    result <-+      catchError+        (runTxOnPool (store ^. #pool) TxSessions.transaction (Compaction.applyCompactionTx manifest))+        ( \_ err -> case err of+            UnexpectedServerError "23503" message -> throwError (CompactionConcurrentMutation (compactionManifestDigest manifest) message)+            _ -> throwError err+        )+    let digest = compactionManifestDigest manifest+    liftIO $ case result of+      Right (CompactionAppliedNow record) -> do+        let report = record ^. #report+            memberships = report ^. #homeMemberships + report ^. #globalMemberships + report ^. #linkMemberships+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionApplied (record ^. #compactionId) digest (report ^. #selectedEvents) memberships)+      Right (CompactionAlreadyApplied record) ->+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionAlreadyApplied (record ^. #compactionId) digest)+      Left refusals ->+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionRefused digest (NonEmpty.head refusals) (NonEmpty.length refusals))+    pure result+  GetCompactionLedger query ->+    runTxOnPool (store ^. #pool) TxSessions.transaction (Compaction.compactionLedgerTx query)+  PreviewCompaction manifest -> do+    result <- runReadOnlyTxOnPool (store ^. #pool) (Compaction.previewCompactionTx manifest)+    let digest = compactionManifestDigest manifest+    liftIO $ case result of+      Right (CompactionPreviewReady report) ->+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionPreviewed digest (report ^. #selectedEvents))+      Right (CompactionPreviewAlreadyApplied record) ->+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionPreviewed digest (record ^. #report . #selectedEvents))+      Left refusals ->+        emitOrDrop (store ^. #eventHandler) (KirokuEventCompactionRefused digest (NonEmpty.head refusals) (NonEmpty.length refusals))+    pure result+  GetStoreIdentity ->+    usePool (store ^. #pool) $ Session.statement () SQL.storeIdentityStmt   GetVisibleGlobalHeadPosition ->     usePool (store ^. #pool) $       Session.statement () SQL.visibleGlobalHeadPositionStmt@@ -278,6 +327,23 @@   EventExistsInStream (StreamName name) (EventId eid) ->     usePool (store ^. #pool) $       Session.statement (name, eid) SQL.eventExistsInStreamStmt+  LookupEventReferences [] ->+    pure Map.empty+  LookupEventReferences eventIds -> do+    let uuids = V.fromList (Map.keys (Map.fromList [(uid, ()) | EventId uid <- eventIds]))+        batches = chunkUuids maxEventReferenceBatch uuids+    usePool (store ^. #pool) $ do+      perBatch <-+        traverse+          ( \batch ->+              (,)+                <$> Session.statement batch SQL.eventMembershipsStmt+                <*> Session.statement batch SQL.deadLetterCountsStmt+          )+          batches+      dependents <- Session.statement uuids SQL.causationDependentCountsStmt+      let (memberships, deadLetters) = unzip perBatch+      pure (assembleEventReferences (V.concat memberships) (V.concat deadLetters) dependents)   LookupStreamNames [] ->     pure Map.empty   LookupStreamNames sids ->@@ -694,3 +760,57 @@ decodeReadEvents settings events = case decodeHook settings of   Nothing -> pure events   Just hook -> V.mapM (\event -> liftIO (hook event) >>= either (throwError . EventDecodeFailed) pure) events++-- | Ids per execution of the membership and dead-letter statements. The+-- causation-dependent statement is exempt: its array is also its exclusion set,+-- so it always runs once over the full request.+maxEventReferenceBatch :: Int+maxEventReferenceBatch = 10000++-- | Split a request into batches of at most 'maxEventReferenceBatch' ids.+chunkUuids :: Int -> Vector UUID -> [Vector UUID]+chunkUuids size values+  | V.null values = []+  | otherwise =+      let (batch, rest) = V.splitAt (max 1 size) values+       in batch : chunkUuids size rest++-- | Fold the three inventory statements' rows into one map. Memberships keep+-- statement order (ascending stream id, so the global row's entry comes first);+-- each row becomes 'GlobalMembership' (@stream_id = 0@, carrying the global+-- position), 'HomeMembership' (@stream_id = original_stream_id@), or+-- 'LinkMembership'.+assembleEventReferences ::+  Vector (UUID, Int64, Text, Int64, Int64, Int64) ->+  Vector (UUID, Int64) ->+  Vector (UUID, Int64) ->+  Map EventId EventReferenceInventory+assembleEventReferences memberships deadLetters dependents =+  Map.mapWithKey attach grouped+  where+    grouped =+      Map.fromListWith+        (<>)+        [ (EventId eid, [membership sid name version origin])+        | (eid, sid, name, version, origin, _originVersion) <- V.toList memberships+        ]+    membership sid name version origin+      | sid == 0 = GlobalMembership (GlobalPosition version)+      | sid == origin = HomeMembership (StreamName name) (StreamId sid) (StreamVersion version)+      | otherwise = LinkMembership (StreamName name) (StreamId sid) (StreamVersion version)+    deadLetterMap = Map.fromList [(EventId eid, n) | (eid, n) <- V.toList deadLetters]+    dependentMap = Map.fromList [(EventId eid, n) | (eid, n) <- V.toList dependents]+    attach eid ms =+      EventReferenceInventory+        { memberships = V.fromList (reverse ms),+          deadLetterCount = Map.findWithDefault 0 eid deadLetterMap,+          causationDependentCount = Map.findWithDefault 0 eid dependentMap+        }++-- | One stable snapshot for maintenance reads, without writes or row locks.+runReadOnlyTxOnPool :: (IOE :> es, Error StoreError :> es) => Pool -> Tx.Transaction a -> Eff es a+runReadOnlyTxOnPool pool tx = do+  result <- liftIO $ Pool.use pool $ TxSessions.transaction TxSessions.RepeatableRead TxSessions.Read tx+  case result of+    Left usageErr -> throwError (mapTransactionUsageError usageErr)+    Right value -> pure value

@@ -33,6 +33,7 @@ import GHC.Generics (Generic) import Hasql.Errors qualified as Errors import Hasql.Pool (UsageError (..))+import Kiroku.Store.Compaction.Types (CompactionDigest, parseCompactionDigestHex) import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict) import Kiroku.Store.Settings (DecodeFailure) import Kiroku.Store.Types@@ -149,6 +150,10 @@     --       human-readable message. This is *not* generally retryable —     --       investigate.     UnexpectedServerError !Text !Text+  | -- | Retryable concurrent reference/count mutation during compaction. The+    -- transaction committed nothing; re-preview and re-apply. Carries the+    -- manifest digest and the guard or foreign-key failure that aborted it.+    CompactionConcurrentMutation !CompactionDigest !Text   | -- | Catch-all for everything not matched by a more specific     --       constructor above. Retained for backward compatibility with     --       consumers that already pattern-match on it; new code should@@ -215,6 +220,9 @@ mapTransactionUsageError :: UsageError -> StoreError mapTransactionUsageError usageErr =   case extractServerError usageErr of+    Just (Errors.ServerError "KRCMP" message (Just detail) _ _)+      | Just digest <- parseCompactionDigestHex detail ->+          CompactionConcurrentMutation digest message     Just (Errors.ServerError "23505" message detail _ _)       | uniqueConstraintName message detail == Just "events_pkey" ->           DuplicateEvent (EventId <$> (detail >>= extractUuidFromDetail))

@@ -53,10 +53,11 @@  import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO) import Data.Foldable (for_)-import Data.Int (Int32)+import Data.Int (Int32, Int64) import Data.Time (NominalDiffTime) import Data.Time.Clock (UTCTime) import Hasql.Pool (UsageError)+import Kiroku.Store.Compaction.Types (CompactionDigest, CompactionId, CompactionRefusal) import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict, HistoryRetentionLeaseId, HistoryRetentionLeaseOwner, HistoryRetentionPruneResult) import Kiroku.Store.Settings (DecodeFailure) import Kiroku.Store.Subscription.Fsm (DeadLetterReason (..), SubscriptionStopReason (..))@@ -226,6 +227,14 @@     KirokuEventHistoryRetentionLeasesPruned !HistoryRetentionPruneResult   | -- | Supported hard delete was refused before changing history.     KirokuEventHardDeleteHistoryRetentionConflict !StreamName !HistoryRetentionConflict+  | -- | Preview completed: manifest digest and selection count; no payload data.+    KirokuEventCompactionPreviewed !CompactionDigest !Int64+  | -- | Expected refusal: digest, first refusal and total refusal count.+    KirokuEventCompactionRefused !CompactionDigest !CompactionRefusal !Int+  | -- | Committed apply: ledger ID, manifest digest, events and memberships removed.+    KirokuEventCompactionApplied !CompactionId !CompactionDigest !Int64 !Int64+  | -- | Recorded manifest recognized; no row changed.+    KirokuEventCompactionAlreadyApplied !CompactionId !CompactionDigest   deriving stock (Show)  -- | Invoke the optional observability handler, dropping any synchronous exception

@@ -5,6 +5,7 @@     readAllForward,     readAllBackward,     visibleGlobalHeadPosition,+    storeIdentity,     readCategory,     getStream,     getStreamWithHead,@@ -15,6 +16,7 @@     eventExistsInStream,     lookupStreamName,     lookupStreamNames,+    lookupEventReferences,   ) where @@ -289,3 +291,32 @@ -- | An event as seen in the global log, using the configured typed decode hook. getEvent :: (HasCallStack, Store :> es) => EventId -> Eff es (Maybe RecordedEvent) getEvent eid = send (GetEvent eid)++-- | Read the immutable store identity in one statement. Restored clones share+-- their source identity; requires migration 0017.+storeIdentity :: (HasCallStack, Store :> es) => Eff es StoreIdentity+storeIdentity = send GetStoreIdentity++-- | Everything that refers to each of the given events. For every id that+-- exists, the result holds its 'EventReferenceInventory': memberships (the+-- @$all@ 'GlobalMembership' first, then home and link entries in ascending+-- stream-id order), the count of @dead_letters@ rows referencing it, and the+-- count of other events whose @causationId@ is it (events in the request are+-- not counted as each other's dependents). Ids that name no event are absent.+--+-- Soft-deleted streams are reported like live ones: the rows physically exist.+-- Use 'getStream' when stream state matters.+--+-- Passing @[]@ returns an empty map without a database round trip. Non-empty+-- input of any size runs on one pooled connection: the membership and+-- dead-letter statements execute in batches of at most 10000 ids, and the+-- causation statement executes once over the full request (its input is also its+-- exclusion set). This is an operator and maintenance read, not a hot path; it+-- is the inventory a caller consults before asking Kiroku to remove events (see+-- "Kiroku.Store.Compaction"). Inventory is advisory under concurrent mutation;+-- preview validates one repeatable-read snapshot and apply revalidates under locks.+lookupEventReferences ::+  (HasCallStack, Store :> es) =>+  [EventId] ->+  Eff es (Map EventId EventReferenceInventory)+lookupEventReferences eventIds = send (LookupEventReferences eventIds)

@@ -35,7 +35,13 @@     getEventStmt,     currentGlobalPositionStmt,     visibleGlobalHeadPositionStmt,+    storeIdentityStmt, +    -- * Event reference inventory statements+    eventMembershipsStmt,+    deadLetterCountsStmt,+    causationDependentCountsStmt,+     -- * Consumer-group read statements     readCategoryForwardConsumerGroupStmt,     readCategoryConsumerGroupEncoder,@@ -1653,3 +1659,77 @@     "SELECT e.event_id,e.event_type,se.stream_version,se.stream_version AS global_position, se.original_stream_id,se.original_stream_version,e.data,e.metadata,e.causation_id,e.correlation_id,e.created_at FROM events e JOIN stream_events se ON se.event_id=e.event_id AND se.stream_id=0 WHERE e.event_id=$1"     (E.param (E.nonNullable E.uuid))     (D.rowMaybe recordedEventRow)++-- | The singleton identity installed by migration 0017.+storeIdentityStmt :: Statement () StoreIdentity+storeIdentityStmt =+  preparable+    "SELECT store_id FROM store_identity WHERE singleton"+    E.noParams+    (D.singleRow (StoreIdentity <$> D.column (D.nonNullable D.uuid)))++-- | Every junction row for the given event ids, with the stream name resolved.+-- Served by @stream_events_pkey@ on @(event_id, stream_id)@; the join to+-- @streams@ is a primary-key lookup per row. Rows are returned in ascending+-- @(event_id, stream_id)@ order so callers can group without sorting.+eventMembershipsStmt :: Statement (Vector UUID) (Vector (UUID, Int64, Text, Int64, Int64, Int64))+eventMembershipsStmt =+  preparable+    """+    SELECT se.event_id, se.stream_id, s.stream_name,+           se.stream_version, se.original_stream_id, se.original_stream_version+    FROM stream_events se+    JOIN streams s ON s.stream_id = se.stream_id+    WHERE se.event_id = ANY($1::uuid[])+    ORDER BY se.event_id, se.stream_id+    """+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    ( D.rowVector+        ( (,,,,,)+            <$> D.column (D.nonNullable D.uuid)+            <*> D.column (D.nonNullable D.int8)+            <*> D.column (D.nonNullable D.text)+            <*> D.column (D.nonNullable D.int8)+            <*> D.column (D.nonNullable D.int8)+            <*> D.column (D.nonNullable D.int8)+        )+    )++-- | Dead-letter rows per referenced event, across all subscriptions and+-- members. Served by @ix_dead_letters_event_id@. Events with no dead letters are+-- absent from the result.+deadLetterCountsStmt :: Statement (Vector UUID) (Vector (UUID, Int64))+deadLetterCountsStmt =+  preparable+    """+    SELECT dl.event_id, count(*)+    FROM dead_letters dl+    WHERE dl.event_id = ANY($1::uuid[])+    GROUP BY dl.event_id+    """+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    (D.rowVector ((,) <$> D.column (D.nonNullable D.uuid) <*> D.column (D.nonNullable D.int8)))++-- | Per referenced event, the number of other events whose @causation_id@ is+-- that event, excluding events that are themselves in the input set. Served by+-- the partial index @ix_events_causation_id@. The exclusion is an anti-join over+-- @unnest($1)@ — never @NOT (event_id = ANY($1))@, which degrades to a linear+-- scan of the array per candidate row at large input sizes. The input array is+-- simultaneously the lookup set and the exclusion set, so callers always execute+-- this statement once over the full request (batching it would silently change+-- "outside the input set" into "outside the batch"). Events with no dependents+-- are absent from the result.+causationDependentCountsStmt :: Statement (Vector UUID) (Vector (UUID, Int64))+causationDependentCountsStmt =+  preparable+    """+    SELECT e.causation_id, count(*)+    FROM events e+    LEFT JOIN unnest($1::uuid[]) AS excluded(event_id)+      ON excluded.event_id = e.event_id+    WHERE e.causation_id = ANY($1::uuid[])+      AND excluded.event_id IS NULL+    GROUP BY e.causation_id+    """+    (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.uuid))))+    (D.rowVector ((,) <$> D.column (D.nonNullable D.uuid) <*> D.column (D.nonNullable D.int8)))

@@ -42,6 +42,8 @@   ( -- * Bare escape hatch     runTransaction,     runTransactionNoRetry,+    storeIdentityTx,+    lookupEventReferencesTx,      -- * Tx-flavored append building block     appendToStreamTx,@@ -67,15 +69,19 @@ import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Functor (($>)) import Data.Generics.Labels ()+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map import Data.Time.Clock (UTCTime, getCurrentTime)+import Data.Vector qualified as V import Effectful (Eff, IOE, (:>)) import Effectful.Dispatch.Dynamic (send) import GHC.Stack (HasCallStack) import Hasql.Transaction qualified as Tx import Kiroku.Store.Connection (KirokuStore)-import Kiroku.Store.Effect (PreparedEvent, Store (..), appendDispatchTx, buildAppendParams, prepareEvents)+import Kiroku.Store.Effect (PreparedEvent, Store (..), appendDispatchTx, assembleEventReferences, buildAppendParams, chunkUuids, maxEventReferenceBatch, prepareEvents) import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore) import Kiroku.Store.Error (AppendConflict (..), StoreError (..), appendConflictToStoreError, emptyResultConflict, validateStreamName)+import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings qualified as Settings import Kiroku.Store.Types @@ -361,3 +367,29 @@               Right ar ->                 Right <$> k ar       send (ctor body)++-- | Read the immutable store identity inside the caller's transaction.+storeIdentityTx :: Tx.Transaction StoreIdentity+storeIdentityTx = Tx.statement () SQL.storeIdentityStmt++-- | 'Kiroku.Store.Read.lookupEventReferences' inside the caller's transaction,+-- with caller-chosen isolation. Use repeatable read when inventory statements+-- and decisions must share one snapshot. Empty input runs no statement. Batching mirrors the interpreter: membership and+-- dead-letter statements at most 10000 ids per execution, the causation+-- statement once over the full request.+lookupEventReferencesTx :: [EventId] -> Tx.Transaction (Map EventId EventReferenceInventory)+lookupEventReferencesTx [] = pure Map.empty+lookupEventReferencesTx eventIds = do+  let uuids = V.fromList (Map.keys (Map.fromList [(uid, ()) | EventId uid <- eventIds]))+      batches = chunkUuids maxEventReferenceBatch uuids+  perBatch <-+    traverse+      ( \batch ->+          (,)+            <$> Tx.statement batch SQL.eventMembershipsStmt+            <*> Tx.statement batch SQL.deadLetterCountsStmt+      )+      batches+  dependents <- Tx.statement uuids SQL.causationDependentCountsStmt+  let (memberships, deadLetters) = unzip perBatch+  pure (assembleEventReferences (V.concat memberships) (V.concat deadLetters) dependents)

@@ -5,6 +5,7 @@     EventType (..),     StreamVersion (..),     GlobalPosition (..),+    StoreIdentity (..),     ExpectedVersion (..),     EventData (..),     RecordedEvent (..),@@ -15,6 +16,8 @@     categoryName,     streamNameInCategory,     EventFilter (..),+    EventMembership (..),+    EventReferenceInventory (..),     BrowsePageSize,     BrowsePageSizeError (..),     mkBrowsePageSize,@@ -22,12 +25,13 @@   ) where -import Data.Aeson (Value)+import Data.Aeson (FromJSON (..), ToJSON (..), Value) import Data.Int (Int32, Int64) import Data.Text (Text) import Data.Text qualified as Text import Data.Time (UTCTime) import Data.UUID (UUID)+import Data.Vector (Vector) import GHC.Generics (Generic)  -- Stream identification@@ -323,3 +327,40 @@  browsePageSizeValue :: BrowsePageSize -> Int32 browsePageSizeValue (BrowsePageSize n) = n++-- | Immutable, migration-installed store identity. Restored clones share the+-- source identity so the same maintenance manifest can be rehearsed on a clone.+newtype StoreIdentity = StoreIdentity UUID+  deriving stock (Eq, Ord, Show, Generic)++-- | How an event is visible in one stream. Every event has exactly one+-- 'HomeMembership' (the stream it was appended to, with the version it holds+-- there) and one 'GlobalMembership' (the @$all@ log, carrying the event's global+-- position); each 'Kiroku.Store.Link.linkToStream' adds one 'LinkMembership'+-- naming the target stream and the version the event holds in it. Each+-- constructor carries exactly the identity its membership class has, so a global+-- position is never smuggled through a 'StreamVersion' field.+data EventMembership+  = HomeMembership !StreamName !StreamId !StreamVersion+  | GlobalMembership !GlobalPosition+  | LinkMembership !StreamName !StreamId !StreamVersion+  deriving stock (Eq, Show, Generic)++-- | Everything in the store that refers to one event: its memberships — the+-- 'GlobalMembership' first (its junction row is stream id 0), then home and link+-- entries in ascending stream-id order — the number of @dead_letters@ rows that+-- reference it (across all subscriptions), and the number of /other/ events+-- whose @causation_id@ is this event. Produced by+-- 'Kiroku.Store.Read.lookupEventReferences'.+data EventReferenceInventory = EventReferenceInventory+  { memberships :: !(Vector EventMembership),+    deadLetterCount :: !Int64,+    causationDependentCount :: !Int64+  }+  deriving stock (Eq, Show, Generic)++instance ToJSON StoreIdentity where+  toJSON (StoreIdentity uid) = toJSON uid++instance FromJSON StoreIdentity where+  parseJSON value = StoreIdentity <$> parseJSON value

@@ -32,11 +32,18 @@ import Test.Category qualified as Category import Test.CategoryIdleNoSpin qualified as CategoryIdleNoSpin import Test.Causation qualified as Causation+import Test.CompactionApply qualified as CompactionApply+import Test.CompactionApplyMock qualified as CompactionApplyMock+import Test.CompactionManifest qualified as CompactionManifest+import Test.CompactionPreview qualified as CompactionPreview+import Test.CompactionPreviewMock qualified as CompactionPreviewMock import Test.Concurrency qualified as Concurrency import Test.ConsumerGroup qualified as ConsumerGroup import Test.ConsumerGroupEffect qualified as ConsumerGroupEffect import Test.ConsumerGroupResize qualified as ConsumerGroupResize import Test.ConsumerGroupSql qualified as ConsumerGroupSql+import Test.EventReferences qualified as EventReferences+import Test.EventReferencesMock qualified as EventReferencesMock import Test.EventTypeFilter qualified as EventTypeFilter import Test.FailureInjection qualified as FailureInjection import Test.HandlerStall qualified as HandlerStall@@ -56,6 +63,8 @@ import Test.ReadStream qualified as ReadStream import Test.ScriptSessionErrors qualified as ScriptSessionErrors import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing+import Test.StoreIdentity qualified as StoreIdentity+import Test.StoreIdentityMock qualified as StoreIdentityMock import Test.StreamBridgeTermination qualified as StreamBridgeTermination import Test.StreamHead qualified as StreamHead import Test.StreamHeadIsolation qualified as StreamHeadIsolation@@ -85,6 +94,15 @@  main :: IO () main = withSharedMigratedPostgres $ hspec $ do+  CompactionApply.spec+  CompactionApplyMock.spec+  CompactionPreview.spec+  CompactionPreviewMock.spec+  CompactionManifest.spec+  EventReferences.spec+  EventReferencesMock.spec+  StoreIdentity.spec+  StoreIdentityMock.spec   BrowseReads.spec   BrowseReadsMock.spec   SubscriptionDeadLetters.spec

@@ -0,0 +1,395 @@+{-# LANGUAGE MultilineStrings #-}+{-# LANGUAGE OverloadedRecordDot #-}++module Test.CompactionApply (spec) where++import Control.Concurrent (myThreadId, threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (forM_, void, when)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.UUID qualified as UUID+import Data.Vector qualified as V+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 (preparable)+import Hasql.Transaction qualified as Tx+import Hedgehog hiding (property)+import Hedgehog.Gen qualified as Gen+import Hedgehog.Range qualified as Range+import Kiroku.Store hiding (id, target)+import Kiroku.Store.SQL qualified as StoreSQL+import Kiroku.Test.Postgres (withMigratedTestDatabase)+import Test.CompactionHelpers+import Test.Helpers+import Test.Hspec+import Test.Hspec.Hedgehog (hedgehog, modifyMaxSuccess)++spec :: Spec+spec = describe "compaction apply" $ do+  it "removes exactly the selections, preserves retained envelopes and records the preview report" $ withTestStore $ \store -> do+    events <- seed store origin 6+    input <- inputFor store [events !! 1, events !! 3]+    let m = manifest input+        kept = filter (\e -> e.eventId `notElem` map (\s -> s.eventId) input.selections) events+    preview <- readyReport store m+    Right (CompactionAppliedNow record) <- storeRead =<< runStoreIO store (applyCompaction m)+    record.report `shouldBe` preview+    runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right (V.fromList kept)+    countEvents store `shouldReturn` 4+    Just info <- storeRead =<< runStoreIO store (getStream origin)+    info.version `shouldBe` StreamVersion 6+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right (V.singleton record)+    runStoreIO store (runTransaction (compactionLedgerTx (ledgerQuery 10))) `shouldReturn` Right (V.singleton record)+    record.report.reportDigest `shouldBe` compactionReportDigest record.report+    runStoreIO store (applyCompaction m) `shouldReturn` Right (Right (CompactionAlreadyApplied record))+    runStoreIO store (previewCompaction m) `shouldReturn` Right (Right (CompactionPreviewAlreadyApplied record))+    let retainedVersions = [1, 3, 5, 6]+    forward <- storeRead =<< runStoreIO store (readStreamForward origin (StreamVersion 0) 100)+    map (\e -> e.streamVersion) (V.toList forward) `shouldBe` map StreamVersion retainedVersions+    backward <- storeRead =<< runStoreIO store (readStreamBackward origin (StreamVersion 100) 100)+    map (\e -> e.streamVersion) (V.toList backward) `shouldBe` map StreamVersion (reverse retainedVersions)+    allBack <- storeRead =<< runStoreIO store (readAllBackward (GlobalPosition 100) 100)+    allBack `shouldBe` V.fromList (reverse kept)+    afterGap <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 2) 100)+    afterGap `shouldBe` V.fromList (filter (\e -> e.globalPosition > GlobalPosition 2) kept)+    category <- storeRead =<< runStoreIO store (readCategory (CategoryName "orders") (GlobalPosition 0) 100)+    category `shouldBe` V.fromList kept+    appended <- storeRead =<< runStoreIO store (appendToStream origin (ExactVersion (StreamVersion 6)) [makeEvent "AfterCompaction" (Aeson.object [])])+    appended.streamVersion `shouldBe` StreamVersion 7+    runStoreIO store (appendToStream origin (ExactVersion (StreamVersion 4)) [makeEvent "Stale" (Aeson.object [])])+      `shouldReturn` Left (WrongExpectedVersion origin (ExactVersion (StreamVersion 4)) (StreamVersion 0))++  it "rolls back both deletion and ledger insertion when the caller condemns its transaction" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [events !! 1]+    Right (CompactionAppliedNow _) <- storeRead =<< runStoreIO store (runTransaction (applyCompactionTx (manifest input) <* Tx.condemn))+    runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right (V.fromList events)+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right V.empty++  it "refuses unexpected links and preserves retained links and all heads" $ withTestStore $ \store -> do+    events <- seed store origin 3+    _ <- storeRead =<< runStoreIO store (linkToStream target (map (\e -> e.eventId) events))+    _ <- storeRead =<< runStoreIO store (linkToStream mirror [(events !! 1).eventId])+    input <- inputFor store [events !! 1]+    let [selection] = input.selections+        refused = manifest (input & #selections .~ [selection & #acknowledgedLinks .~ V.empty] & #streamHeads .~ [StreamHeadWitness origin (StreamVersion 3)])+    Right (Left _) <- runStoreIO store (applyCompaction refused)+    countEvents store `shouldReturn` 3+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right V.empty+    before <- storeRead =<< runStoreIO store (readStreamForward target (StreamVersion 0) 100)+    Right (CompactionAppliedNow record) <- storeRead =<< runStoreIO store (applyCompaction (manifest input))+    record.report.linkMemberships `shouldBe` 2+    after <- storeRead =<< runStoreIO store (readStreamForward target (StreamVersion 0) 100)+    after `shouldBe` V.filter (\e -> e.eventId /= selection.eventId) before+    Just info <- storeRead =<< runStoreIO store (getStream target)+    info.version `shouldBe` StreamVersion 3++  it "retains high-water frontiers after tail and complete-stream compaction" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store (drop 1 events)+    _ <- applied store (manifest input)+    runStoreIO store visibleGlobalHeadPosition `shouldReturn` Right (GlobalPosition 1)+    frontier <- storeRead =<< runStoreIO store subscriptionCheckpointInventory+    frontier.storePosition `shouldBe` GlobalPosition 3+    lastInput <- inputFor store [head events]+    _ <- applied store (manifest lastInput)+    runStoreIO store visibleGlobalHeadPosition `shouldReturn` Right (GlobalPosition 0)+    Just info <- storeRead =<< runStoreIO store (getStream origin)+    info.version `shouldBe` StreamVersion 3+    append <- storeRead =<< runStoreIO store (appendToStream origin (ExactVersion (StreamVersion 3)) [makeEvent "AfterEmpty" (Aeson.object [])])+    (append.streamVersion, append.globalPosition) `shouldBe` (StreamVersion 4, GlobalPosition 4)++  it "refuses changed heads, missing events and a ledger whose IDs reappear" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [events !! 1]+    _ <- seed store origin 1+    runStoreIO store (applyCompaction (manifest input))+      `shouldReturn` Right (Left (CompactionStreamHeadDrift origin (StreamVersion 3) (StreamVersion 4) :| []))+    fresh <- inputFor store [events !! 1]+    record <- applied store (manifest fresh)+    let missing = manifest (fresh & #operation .~ operation "never-applied")+    runStoreIO store (applyCompaction missing) `shouldReturn` Right (Left (CompactionSelectedEventMissing (events !! 1).eventId :| []))+    _ <- storeRead =<< runStoreIO store (appendToStream (StreamName "reused-id") NoStream [makeEvent "Reused" (Aeson.object []) & #eventId .~ Just (events !! 1).eventId])+    runStoreIO store (applyCompaction (manifest fresh)) `shouldReturn` Right (Left (CompactionLedgerConflict record.compactionId 1 :| []))++  it "refuses a partially matching store before deleting any surviving selection" $ withTestStore $ \store -> do+    events <- seed store origin 3+    whole <- inputFor store (take 2 events)+    first <- inputFor store [head events]+    _ <- applied store (manifest first)+    runStoreIO store (applyCompaction (manifest whole))+      `shouldReturn` Right (Left (CompactionSelectedEventMissing (head events).eventId :| []))+    runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right (V.fromList (tail events))+    records <- storeRead =<< runStoreIO store (compactionLedger (ledgerQuery 10))+    V.length records `shouldBe` 1++  it "removes dead letters only by policy and permits explicitly dangling causation" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    let EventId uid = event.eventId+    _ <- storeRead =<< runStoreIO store (appendToStream target NoStream [makeEvent "Dependent" (Aeson.object []) & #causationId .~ Just uid])+    insertDeadLetterForEvent store "projection" event+    input <- inputFor store [event]+    Right (Left _) <- runStoreIO store (applyCompaction (manifest input))+    countEvents store `shouldReturn` 2+    record <- applied store (manifest (input & #deadLetterPolicy .~ RemoveDeadLetters & #causationPolicy .~ AllowDanglingCausation))+    (record.report.deadLettersRemoved, record.report.causationDependents) `shouldBe` (1, 1)+    countDeadLettersForEvents store [EventId uid] `shouldReturn` 0+    kept <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+    (V.head kept).causationId `shouldBe` Just uid++  it "refuses an active lease, then applies below a logical truncate marker" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [head events]+    _ <- storeRead =<< runStoreIO store (setStreamTruncateBefore origin (StreamVersion 3))+    let Right owner = mkHistoryRetentionLeaseOwner "apply"+        Right reason = mkHistoryRetentionLeaseReason "test"+        Right duration = mkHistoryRetentionLeaseDuration 60+    lease <- storeRead =<< runStoreIO store (acquireHistoryRetentionLease (HistoryRetentionLeaseRequest owner reason duration))+    Right (Left (CompactionHistoryRetentionActive _ :| [])) <- runStoreIO store (applyCompaction (manifest input))+    countEvents store `shouldReturn` 3+    _ <- storeRead =<< runStoreIO store (releaseHistoryRetentionLease (HistoryRetentionLeaseHandle lease.leaseId owner))+    _ <- applied store (manifest input)+    Just info <- storeRead =<< runStoreIO store (getStream origin)+    info.truncateBefore `shouldBe` StreamVersion 3++  forM_ [(AllStreams, 1, False), (AllStreams, 2, False), (Category (CategoryName "orders"), 2, False), (AllStreams, 2, True), (Category (CategoryName "orders"), 2, True)] $ \(subscriptionTarget, checkpoint, grouped) ->+    it ("continues subscription catch-up and live delivery across removed positions: " <> show (subscriptionTarget, checkpoint, grouped)) $ do+      ready <- newEmptyMVar+      let name = SubscriptionName "compaction-resume"+          tweak settings = settings & #eventHandler .~ Just (caughtUpEventHandler name ready Nothing)+      withTestStoreSettings tweak $ \store -> do+        events <- seed store origin 6+        input <- inputFor store [events !! 1, events !! 3]+        _ <- applied store (manifest input)+        _ <- storeRead =<< runStoreIO store (runTransaction (Tx.statement ("compaction-resume", checkpoint) StoreSQL.saveCheckpointStmt))+        received <- newIORef []+        let handler event = do+              modifyIORef' received (event.globalPosition :)+              pure (if event.globalPosition == GlobalPosition 7 then Stop else Continue)+            config = (defaultSubscriptionConfig name subscriptionTarget handler) {consumerGroup = if grouped then Just (validConsumerGroup 0 1) else Nothing}+        handle <- subscribe store config+        within (waitForSubscriptionLive ready)+        _ <- storeRead =<< runStoreIO store (appendToStream origin (ExactVersion (StreamVersion 6)) [makeEvent "PostGap" (Aeson.object [])])+        completed <- waitWithTimeout 5_000_000 handle+        completed `shouldSatisfy` \case Right (Right ()) -> True; _ -> False+        reverse <$> readIORef received `shouldReturn` map GlobalPosition [3, 5, 6, 7]+        inventory <- storeRead =<< runStoreIO store subscriptionCheckpointInventory+        -- The worker persists the terminal delivery through the normal save path.+        rows <- storeRead =<< runStoreIO store (runTransaction (Tx.statement ("compaction-resume", 0) StoreSQL.getCheckpointMemberStmt))+        rows `shouldBe` Just 7+        inventory.storePosition `shouldBe` GlobalPosition 7++  it "serializes competing applies into one commit and one identical ledgered no-op" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [head events]+    (first, second) <- Async.concurrently (runStoreIO store (applyCompaction (manifest input))) (runStoreIO store (applyCompaction (manifest input)))+    let records = [record | Right (Right (CompactionAppliedNow record)) <- [first, second]]+        repeated = [record | Right (Right (CompactionAlreadyApplied record)) <- [first, second]]+    length records `shouldBe` 1+    repeated `shouldBe` records++  it "blocks an append while apply holds stream locks, then continues from the unchanged head" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [head events]+    Async.withAsync (runStoreIO store (runTransaction (applyCompactionTx (manifest input) <* pause "apply-held"))) $ \worker -> do+      waitPhase store "apply-held"+      Async.withAsync (runStoreIO store (appendToStream origin (ExactVersion (StreamVersion 3)) [makeEvent "BlockedAppend" (Aeson.object [])])) $ \writer -> do+        threadDelay 50_000+        Async.poll writer >>= (`shouldSatisfy` maybe True (const False))+        Right (Right (CompactionAppliedNow _)) <- within (Async.wait worker)+        Right result <- within (Async.wait writer)+        result.streamVersion `shouldBe` StreamVersion 4++  it "waits for a replay stream guard, but a recorded no-op never takes stream locks" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [head events]+    let m = manifest input+    Async.withAsync (runStoreIO store (runTransaction (lockStreamHistoryForReplayTx origin <* pause "guard-held"))) $ \guard -> do+      waitPhase store "guard-held"+      Async.withAsync (runStoreIO store (applyCompaction m)) $ \worker -> do+        threadDelay 50_000+        Async.poll worker >>= (`shouldSatisfy` maybe True (const False))+        _ <- within (Async.wait guard)+        Right (Right (CompactionAppliedNow _)) <- within (Async.wait worker)+        pure ()+    Async.withAsync (runStoreIO store (runTransaction (lockStreamHistoryForReplayTx origin <* pause "noop-guard"))) $ \guard -> do+      waitPhase store "noop-guard"+      Right (Right (CompactionAlreadyApplied _)) <- withinShort (runStoreIO store (applyCompaction m))+      void (within (Async.wait guard))++  it "rejects another store's manifest before queuing for the coordinator" $ withTestStore $ \store -> do+    events <- seed store origin 1+    input <- inputFor store events+    let wrong = manifest (input & #storeIdentity .~ StoreIdentity UUID.nil)+    Async.withAsync (runStoreIO store (runTransaction (Tx.sql "SELECT singleton FROM history_retention_coordinator WHERE singleton FOR UPDATE" >> pause "coordinator-held"))) $ \holder -> do+      waitPhase store "coordinator-held"+      withinShort (runStoreIO store (applyCompaction wrong))+        `shouldReturn` Right (Left (CompactionStoreIdentityMismatch (StoreIdentity UUID.nil) input.storeIdentity :| []))+      void (within (Async.wait holder))++  it "keeps an unlisted concurrent link and rolls back every deletion (direct composer FK path)" $ withTestStore $ \store -> do+    events <- seed store origin 2+    input <- inputFor store [head events]+    Async.withAsync (runStoreIO store (runTransaction (applyCompactionTxWith (pause "concurrent-link") (manifest input)))) $ \worker -> do+      waitPhase store "concurrent-link"+      _ <- storeRead =<< runStoreIO store (linkToStream target [(head events).eventId])+      result <- within (Async.wait worker)+      result `shouldSatisfy` \case Left (UnexpectedServerError "23503" _) -> True; _ -> False+    runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right (V.fromList events)+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right V.empty+    links <- storeRead =<< runStoreIO store (readStreamForward target (StreamVersion 0) 100)+    V.length links `shouldBe` 1++  forM_ [RefuseDeadLetters, RemoveDeadLetters] $ \policy ->+    it ("rolls back a concurrent dead letter under " <> show policy) $ withTestStore $ \store -> do+      events <- seed store origin 2+      input <- inputFor store [head events]+      let m = manifest (input & #deadLetterPolicy .~ policy)+      Async.withAsync (runStoreIO store (runTransaction (applyCompactionTxWith (pause "concurrent-dead-letter") m))) $ \worker -> do+        waitPhase store "concurrent-dead-letter"+        insertDeadLetterForEvent store "concurrent" (head events)+        result <- within (Async.wait worker)+        case policy of+          RefuseDeadLetters -> result `shouldSatisfy` \case Left (UnexpectedServerError "23503" _) -> True; _ -> False+          RemoveDeadLetters -> result `shouldSatisfy` \case Left (CompactionConcurrentMutation digest "dead-letter count") -> digest == compactionManifestDigest m; _ -> False+      countEvents store `shouldReturn` 2+      let EventId uid = (head events).eventId+      countDeadLettersForEvents store [EventId uid] `shouldReturn` 1+      runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right V.empty++  it "maps the real effect's concurrent foreign-key failure to a typed retryable error" $ withTestStore $ \store -> do+    events <- seed store origin 2+    input <- inputFor store [head events]+    raw store "CREATE FUNCTION kiroku.test_compaction_pause() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(0.6); RETURN NULL; END $$; CREATE TRIGGER test_compaction_pause BEFORE DELETE ON kiroku.stream_events FOR EACH STATEMENT EXECUTE FUNCTION kiroku.test_compaction_pause()"+    Async.withAsync (runStoreIO store (applyCompaction (manifest input))) $ \worker -> do+      waitUntil store "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND state='active' AND wait_event='PgSleep' AND query LIKE 'DELETE FROM stream_events%')"+      _ <- storeRead =<< runStoreIO store (linkToStream target [(head events).eventId])+      result <- within (Async.wait worker)+      result `shouldSatisfy` \case Left (CompactionConcurrentMutation digest _) -> digest == compactionManifestDigest (manifest input); _ -> False+    countEvents store `shouldReturn` 2+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right V.empty++  it "documents causation refusal as validation-time best-effort for unlisted writers" $ withTestStore $ \store -> do+    events <- seed store origin 2+    input <- inputFor store [head events]+    let EventId uid = (head events).eventId+    Async.withAsync (runStoreIO store (runTransaction (applyCompactionTxWith (pause "concurrent-causation") (manifest input)))) $ \worker -> do+      waitPhase store "concurrent-causation"+      _ <- storeRead =<< runStoreIO store (appendToStream target NoStream [makeEvent "LateDependent" (Aeson.object []) & #causationId .~ Just uid])+      Right (Right (CompactionAppliedNow record)) <- within (Async.wait worker)+      record.report.causationDependents `shouldBe` 0+    kept <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+    (V.last kept).causationId `shouldBe` Just uid++  it "accepts sequential disjoint chunks without changing their shared stream-head witness" $ withTestStore $ \store -> do+    events <- seed store origin 4+    first <- inputFor store [events !! 0]+    second <- inputFor store [events !! 1]+    older <- applied store (manifest first)+    newer <- applied store (manifest second)+    runStoreIO store (compactionLedger (ledgerQuery 1)) `shouldReturn` Right (V.singleton newer)+    runStoreIO store (compactionLedger (ledgerQuery 10)) `shouldReturn` Right (V.fromList [newer, older])+    retained <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+    retained `shouldBe` V.fromList (drop 2 events)++  it "shares KRCMP error mapping with every transaction composer and rejects malformed details" $ withTestStore $ \store -> do+    events <- seed store origin 1+    input <- inputFor store events+    let digest = compactionManifestDigest (manifest input)+        raise detail = runStoreIO store (runTransaction (Tx.sql (TE.encodeUtf8 ("DO $$ BEGIN RAISE EXCEPTION USING ERRCODE='KRCMP', MESSAGE='events count', DETAIL='" <> detail <> "'; END $$"))))+    raise (compactionDigestHex digest) `shouldReturn` Left (CompactionConcurrentMutation digest "events count")+    raise "malformed" `shouldReturn` Left (UnexpectedServerError "KRCMP" "events count")++  it "emits committed, already-applied and refused events with one caller checkout per apply" $ do+    owner <- myThreadId+    calls <- newIORef (0 :: Int)+    eventsRef <- newIORef []+    let tweak settings =+          settings+            { eventHandler = Just (\event -> modifyIORef' eventsRef (event :)),+              observationHandler = Just $ \case+                ConnectionObservation _ InUseConnectionStatus -> do current <- myThreadId; when (current == owner) (modifyIORef' calls (+ 1))+                _ -> pure ()+            }+    withTestStoreSettings tweak $ \store -> do+      events <- seed store origin 2+      input <- inputFor store [head events]+      writeIORef eventsRef []+      before <- readIORef calls+      record <- applied store (manifest input)+      readIORef calls `shouldReturn` before + 1+      emitted <- readIORef eventsRef+      [(cid, digest, count, memberships)] <- pure [(cid, d, n, ms) | KirokuEventCompactionApplied cid d n ms <- emitted]+      (cid, digest, count, memberships) `shouldBe` (record.compactionId, compactionManifestDigest (manifest input), 1, 2)+      _ <- storeRead =<< runStoreIO store (applyCompaction (manifest input))+      again <- readIORef eventsRef+      [repeated] <- pure [cid | KirokuEventCompactionAlreadyApplied cid _ <- again]+      repeated `shouldBe` record.compactionId+      let changed = manifest (input & #operation .~ operation "unrecorded")+      _ <- storeRead =<< runStoreIO store (applyCompaction changed)+      refused <- readIORef eventsRef+      [total] <- pure [n | KirokuEventCompactionRefused _ _ n <- refused]+      total `shouldBe` 1++  modifyMaxSuccess (const 20) $ it "compactingARandomSubsetPreservesTheRest across streams and links" $ hedgehog $ do+    widths <- forAll (Gen.list (Range.linear 2 4) (Gen.int (Range.linear 3 8)))+    picks <- forAll (Gen.list (Range.singleton (sum widths)) Gen.bool)+    evalIO $ withMigratedTestDatabase $ \conn -> withStore (defaultConnectionSettings conn) $ \store -> do+      groups <- sequence [seed store (StreamName ("random-" <> T.pack (show n))) width | (n, width) <- zip [1 :: Int ..] widths]+      let allEvents = concat groups+          chosen = case [e | (e, True) <- zip allEvents picks] of [] -> [head allEvents]; xs -> xs+      _ <- storeRead =<< runStoreIO store (linkToStream target (map (\e -> e.eventId) (take 3 allEvents)))+      input <- inputFor store chosen+      before <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+      linked <- storeRead =<< runStoreIO store (readStreamForward target (StreamVersion 0) 100)+      let selected = map (\e -> e.eventId) chosen+      _ <- applied store (manifest input)+      runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right (V.filter (\e -> e.eventId `notElem` selected) before)+      runStoreIO store (readStreamForward target (StreamVersion 0) 100) `shouldReturn` Right (V.filter (\e -> e.eventId `notElem` selected) linked)+      forM_ input.streamHeads $ \h -> do+        Just stream <- storeRead =<< runStoreIO store (getStream h.stream)+        stream.version `shouldBe` h.headVersion+      countEvents store `shouldReturn` fromIntegral (length allEvents - length chosen)++origin, target, mirror :: StreamName+origin = StreamName "orders-1"+target = StreamName "audit-1"+mirror = StreamName "audit-2"++operation :: Text -> CompactionOperation+operation = either (error . show) id . mkCompactionOperation++applied :: KirokuStore -> CompactionManifest -> IO CompactionRecord+applied store m = do+  Right (CompactionAppliedNow record) <- storeRead =<< runStoreIO store (applyCompaction m)+  pure record++pause :: Text -> Tx.Transaction ()+pause marker = void (Tx.statement () (preparable ("SELECT pg_sleep(0.6) IS NULL /* " <> marker <> " */") E.noParams (D.singleRow (D.column (D.nonNullable D.bool)))))++waitPhase :: KirokuStore -> Text -> IO ()+waitPhase store marker = waitUntil store ("SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND state='active' AND query='SELECT pg_sleep(0.6) IS NULL /* " <> marker <> " */')")++waitUntil :: KirokuStore -> Text -> IO ()+waitUntil store query = within $ loop (150 :: Int)+  where+    loop 0 = fail "compaction barrier was not reached"+    loop n = do+      found <- storeRead =<< Pool.use (store ^. #pool) (Session.statement () (preparable query E.noParams (D.singleRow (D.column (D.nonNullable D.bool)))))+      if found then pure () else threadDelay 10_000 >> loop (n - 1)++within :: IO a -> IO a+within action = Async.race (threadDelay 5_000_000) action >>= either (const (fail "compaction test deadline expired")) pure++withinShort :: IO a -> IO a+withinShort action = Async.race (threadDelay 250_000) action >>= either (const (fail "no-op or identity refusal waited for an unrelated lock")) pure

@@ -0,0 +1,41 @@+module Test.CompactionApplyMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Time (UTCTime (..), fromGregorian)+import Data.UUID qualified as UUID+import Data.Vector qualified as V+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store hiding (id)+import Test.CompactionHelpers (ledgerQuery)+import Test.Hspec++spec :: Spec+spec = describe "compaction apply mock" $+  it "dispatches apply and ledger once each with their original arguments" $ do+    calls <- newIORef []+    let Right op = mkCompactionOperation "mock"+        identity = StoreIdentity UUID.nil+        heads = [StreamHeadWitness (StreamName "mock-1") (StreamVersion 1)]+        selection = CompactionSelection (EventId UUID.nil) (StreamName "mock-1") (StreamVersion 1) (GlobalPosition 1) V.empty+        Right m = mkCompactionManifest (CompactionManifestInput identity op RefuseDeadLetters RefuseCausationDependents heads [selection] Nothing)+        digest = compactionManifestDigest m+        report = sealCompactionReport (CompactionReport digest identity op RefuseDeadLetters RefuseCausationDependents 1 1 1 0 0 0 (GlobalPosition 1) (GlobalPosition 1) (V.fromList heads) digest)+        record = CompactionRecord (CompactionId UUID.nil) report (UTCTime (fromGregorian 2026 10 27) 0) "mock"+        query = ledgerQuery 10+    actual <- runEff $ runApplyMock calls m query record $ (,) <$> applyCompaction m <*> compactionLedger query+    actual `shouldBe` (Right (CompactionAppliedNow record), V.singleton record)+    readIORef calls `shouldReturn` ["ledger", "apply"]++runApplyMock :: (IOE :> es) => IORef [String] -> CompactionManifest -> CompactionLedgerQuery -> CompactionRecord -> Eff (Store : es) a -> Eff es a+runApplyMock calls expected query record = interpret_ $ \case+  ApplyCompaction m -> do+    liftIO $ m `shouldBe` expected+    liftIO $ modifyIORef' calls ("apply" :)+    pure (Right (CompactionAppliedNow record))+  GetCompactionLedger got -> do+    liftIO $ got `shouldBe` query+    liftIO $ modifyIORef' calls ("ledger" :)+    pure (V.singleton record)+  _ -> error "unexpected Store operation in compaction apply mock"

@@ -0,0 +1,58 @@+{-# LANGUAGE OverloadedRecordDot #-}++module Test.CompactionHelpers (seed, inputFor, manifest, ledgerQuery, raw, storeRead, readyReport) where++import Control.Lens ((^.))+import Control.Monad (forM)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.Int (Int32)+import Data.Map.Strict qualified as Map+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Vector qualified as V+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Kiroku.Store hiding (id)+import Test.Helpers (makeEvent)++seed :: KirokuStore -> StreamName -> Int -> IO [RecordedEvent]+seed store name count = do+  _ <- storeRead =<< runStoreIO store (appendToStream name AnyVersion [makeEvent "CompactionFixture" (Aeson.object ["ordinal" Aeson..= n]) | n <- [1 .. count]])+  Just sid <- storeRead =<< runStoreIO store (lookupStreamId name)+  events <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 1000000)+  pure (filter (\event -> event.originalStreamId == sid) (V.toList events))++inputFor :: KirokuStore -> [RecordedEvent] -> IO CompactionManifestInput+inputFor store events = do+  identity <- storeRead =<< runStoreIO store storeIdentity+  references <- storeRead =<< runStoreIO store (lookupEventReferences (map (\e -> e.eventId) events))+  selections <- forM events $ \event -> do+    let inventory = references Map.! event.eventId+        [HomeMembership name _ version] = [home | home@HomeMembership {} <- V.toList inventory.memberships]+        [GlobalMembership position] = [global | global@GlobalMembership {} <- V.toList inventory.memberships]+        links = V.fromList [LinkWitness target version | LinkMembership target _ version <- V.toList inventory.memberships]+    pure (CompactionSelection event.eventId name version position links)+  let names = Set.toList (Set.fromList (concatMap (\s -> s.originStream : map (\l -> l.stream) (V.toList s.acknowledgedLinks)) selections))+  heads <- forM names $ \name -> do+    Just stream <- storeRead =<< runStoreIO store (getStream name)+    pure (StreamHeadWitness name (stream ^. #version))+  operation <- either (fail . show) pure (mkCompactionOperation "test-compaction")+  pure (CompactionManifestInput identity operation RefuseDeadLetters RefuseCausationDependents heads selections Nothing)++manifest :: CompactionManifestInput -> CompactionManifest+manifest = either (error . show) id . mkCompactionManifest++ledgerQuery :: Int32 -> CompactionLedgerQuery+ledgerQuery = CompactionLedgerQuery . either (error . show) id . mkCompactionLedgerLimit++raw :: KirokuStore -> Text -> IO ()+raw store sql = storeRead =<< Pool.use (store ^. #pool) (Session.script sql)++storeRead :: (Show e) => Either e a -> IO a+storeRead = either (fail . show) pure++readyReport :: KirokuStore -> CompactionManifest -> IO CompactionReport+readyReport store m = do+  Right (CompactionPreviewReady report) <- storeRead =<< runStoreIO store (previewCompaction m)+  pure report

@@ -0,0 +1,235 @@+{-# LANGUAGE OverloadedRecordDot #-}++module Test.CompactionManifest (spec) where++import Control.Lens ((&), (.~))+import Control.Monad (forM_)+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString qualified as BS+import Data.Either (isLeft)+import Data.Generics.Labels ()+import Data.Maybe (fromJust)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time (UTCTime (..), fromGregorian)+import Data.UUID qualified as UUID+import Data.Vector qualified as V+import Hedgehog hiding (property)+import Hedgehog.Gen qualified as Gen+import Hedgehog.Range qualified as Range+import Kiroku.Store.Compaction.Types+import Kiroku.Store.HistoryRetention.Types (HistoryRetentionConflict (..))+import Kiroku.Store.Types (EventId (..), GlobalPosition (..), StoreIdentity (..), StreamName (..), StreamVersion (..))+import Test.Hspec+import Test.Hspec.Hedgehog (hedgehog, modifyMaxSuccess)++spec :: Spec+spec = describe "compaction manifest" $ do+  describe "validation" $ do+    it "rejects empty and oversized selections" $ do+      mkCompactionManifest (base & #selections .~ []) `shouldBe` Left CompactionManifestEmpty+      mkCompactionManifest (base & #selections .~ replicate (maxCompactionSelections + 1) selection)+        `shouldBe` Left (CompactionManifestTooLarge (maxCompactionSelections + 1))+    it "rejects duplicate event IDs before duplicate positions" $+      mkCompactionManifest (base & #selections .~ [selection, selection]) `shouldBe` Left (CompactionManifestDuplicateEvent selection.eventId)+    it "rejects duplicate global positions" $+      mkCompactionManifest (base & #selections .~ [selection, selection & #eventId .~ eid 2 & #originVersion .~ StreamVersion 4])+        `shouldBe` Left (CompactionManifestDuplicatePosition (GlobalPosition 42))+    it "rejects duplicate originating stream slots" $+      mkCompactionManifest (base & #selections .~ [selection, selection & #eventId .~ eid 2 & #globalPosition .~ GlobalPosition 43])+        `shouldBe` Left (CompactionManifestDuplicateOriginVersion origin (StreamVersion 3))+    it "rejects reserved origin, link and head names" $ do+      mkCompactionManifest (changeSelection (#originStream .~ reserved)) `shouldBe` Left (CompactionManifestReservedStream reserved)+      mkCompactionManifest (changeSelection (#acknowledgedLinks .~ V.singleton (LinkWitness reserved (StreamVersion 1)))) `shouldBe` Left (CompactionManifestReservedStream reserved)+      mkCompactionManifest (base & #streamHeads .~ [StreamHeadWitness reserved (StreamVersion 1)]) `shouldBe` Left (CompactionManifestReservedStream reserved)+    it "measures names and operation labels in UTF-8 bytes" $ do+      let longName = StreamName (T.replicate 257 "ψ")+      mkCompactionManifest (changeSelection (#originStream .~ longName)) `shouldBe` Left (CompactionManifestInvalidStreamName longName 514)+      mkCompactionOperation "" `shouldBe` Left CompactionManifestOperationEmpty+      mkCompactionOperation (T.replicate 513 "x") `shouldBe` Left (CompactionManifestOperationTooLong 513)+    it "rejects nonpositive origin versions, link versions and global positions" $ do+      mkCompactionManifest (changeSelection (#originVersion .~ StreamVersion 0)) `shouldBe` Left (CompactionManifestNonPositiveVersion selection.eventId)+      mkCompactionManifest (changeSelection (#globalPosition .~ GlobalPosition 0)) `shouldBe` Left (CompactionManifestNonPositivePosition selection.eventId)+      mkCompactionManifest (changeSelection (#acknowledgedLinks .~ V.singleton (LinkWitness target (StreamVersion 0)))) `shouldBe` Left (CompactionManifestNonPositiveVersion selection.eventId)+    it "rejects duplicate links and links into the origin" $ do+      let link = LinkWitness target (StreamVersion 1)+      mkCompactionManifest (changeSelection (#acknowledgedLinks .~ V.fromList [link, link])) `shouldBe` Left (CompactionManifestDuplicateLink selection.eventId link)+      mkCompactionManifest (changeSelection (#acknowledgedLinks .~ V.singleton (LinkWitness origin (StreamVersion 1)))) `shouldBe` Left (CompactionManifestLinkIntoOrigin selection.eventId)+    it "requires exactly one head for every touched stream" $ do+      mkCompactionManifest (base & #streamHeads .~ []) `shouldBe` Left (CompactionManifestMissingStreamHead origin)+      mkCompactionManifest (base & #streamHeads .~ [headWitness, headWitness]) `shouldBe` Left (CompactionManifestDuplicateStreamHead origin)+      mkCompactionManifest (base & #streamHeads .~ [headWitness, StreamHeadWitness target (StreamVersion 1)]) `shouldBe` Left (CompactionManifestUnusedStreamHead target)+    it "refuses an expected digest mismatch" $ do+      let computed = compactionManifestDigest (manifest base)+      mkCompactionManifest (base & #expectedDigest .~ Just zeroDigest)+        `shouldBe` Left (CompactionManifestDigestMismatch zeroDigest computed)+    it "bounds ledger queries and validates digest representations" $ do+      forM_ [0, 1001] $ \n -> mkCompactionLedgerLimit n `shouldBe` Left (CompactionLedgerLimitOutOfRange n)+      mkCompactionDigest (BS.replicate 31 0) `shouldBe` Nothing+      parseCompactionDigestHex (T.replicate 63 "0") `shouldBe` Nothing+      parseCompactionDigestHex (T.replicate 64 "z") `shouldBe` Nothing+      parseCompactionDigestHex (T.replicate 64 "Ｆ") `shouldBe` Nothing+      parseCompactionDigestHex (compactionDigestHex zeroDigest) `shouldBe` Just zeroDigest++  describe "canonical encoding" $ do+    it "pins the independent 134-byte worked example" $ do+      let m = manifest base+      BS.length (canonicalCompactionManifestBytes m) `shouldBe` 134+      canonicalCompactionManifestBytes m+        `shouldBe` BS.pack+          ( [107, 105, 114, 111, 107, 117, 45, 99, 111, 109, 112, 97, 99, 116, 105, 111, 110, 45, 109, 97, 110, 105, 102, 101, 115, 116, 47, 49, 10]+              <> [0, 0, 0, 0, 0, 0, 112, 0, 128, 0, 0, 0, 0, 0, 0, 1]+              <> [0, 0, 0, 7, 99, 111, 109, 112, 97, 99, 116, 0, 0, 0, 0, 0, 1]+              <> [0, 0, 0, 8, 111, 114, 100, 101, 114, 115, 45, 49, 0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 1]+              <> [0, 0, 0, 0, 0, 0, 112, 0, 128, 0, 0, 0, 0, 0, 0, 170]+              <> [0, 0, 0, 8, 111, 114, 100, 101, 114, 115, 45, 49, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 42, 0, 0, 0, 0]+          )+      compactionDigestHex (compactionManifestDigest m) `shouldBe` "965e364734a98abdc509ceb0b4f082ab7a2dbd66713d56ce77d2071e55589d08"+    it "requires format and digest on JSON input" $ do+      let Aeson.Object object = Aeson.toJSON (manifest base)+      (Aeson.fromJSON (Aeson.Object (KM.delete "digest" object)) :: Aeson.Result CompactionManifest) `shouldSatisfy` isError+      (Aeson.fromJSON (Aeson.Object (KM.insert "format" (Aeson.String "wrong") object)) :: Aeson.Result CompactionManifest) `shouldSatisfy` isError+    it "rejects a corrupted report digest" $+      (Aeson.eitherDecode (Aeson.encode (reportFor base & #reportDigest .~ zeroDigest)) :: Either String CompactionReport) `shouldSatisfy` isLeft+    it "round-trips every refusal constructor" $ do+      let now = UTCTime (fromGregorian 2026 10 27) 0+          link = LinkWitness target (StreamVersion 1)+          identity = base.storeIdentity+          refusals =+            [ CompactionStoreIdentityMismatch identity identity,+              CompactionHistoryRetentionActive (HistoryRetentionConflict 2 now),+              CompactionStreamMissing origin,+              CompactionStreamSoftDeleted origin,+              CompactionStreamHeadDrift origin (StreamVersion 1) (StreamVersion 2),+              CompactionSelectedEventMissing selection.eventId,+              CompactionWitnessMismatch selection.eventId (WitnessMismatch origin (StreamVersion 1) Nothing),+              CompactionUnexpectedLink selection.eventId link,+              CompactionAcknowledgedLinkMissing selection.eventId link,+              CompactionDeadLettersPresent selection.eventId 2,+              CompactionCausationDependentsPresent selection.eventId 3,+              CompactionLedgerConflict (CompactionId UUID.nil) 4+            ]+      forM_ refusals $ \r -> Aeson.eitherDecode (Aeson.encode r) `shouldBe` Right r++  modifyMaxSuccess (const 200) $ describe "properties" $ do+    it "is deterministic and insensitive to enumeration order" $ hedgehog $ do+      input <- forAll genInput+      shuffledSelections <- forAll (Gen.shuffle input.selections)+      shuffledHeads <- forAll (Gen.shuffle input.streamHeads)+      shuffledLinks <- traverse (\s -> do ls <- forAll (Gen.shuffle (V.toList s.acknowledgedLinks)); pure (s & #acknowledgedLinks .~ V.fromList ls)) shuffledSelections+      let first = manifest input+          second = manifest (input & #selections .~ shuffledLinks & #streamHeads .~ shuffledHeads)+      compactionManifestDigest first === compactionManifestDigest (manifest input)+      canonicalCompactionManifestBytes first === canonicalCompactionManifestBytes second+    it "changes its digest for every witnessed input field" $ hedgehog $ do+      input <- forAll genInput+      let m = manifest input+          s : ss = input.selections+          h : hs = input.streamHeads+          swapped = StreamName "changed-origin"+          variants =+            [ input & #operation .~ operation "changed",+              input & #deadLetterPolicy .~ RemoveDeadLetters,+              input & #causationPolicy .~ AllowDanglingCausation,+              input & #storeIdentity .~ StoreIdentity UUID.nil,+              input & #streamHeads .~ ((h & #headVersion .~ StreamVersion 9000000) : hs),+              input & #selections .~ ((s & #originVersion .~ StreamVersion 9000000) : ss),+              input & #selections .~ ((s & #globalPosition .~ GlobalPosition 9000000) : ss),+              input & #selections .~ ((s & #eventId .~ eid 9000000) : ss),+              input+                & #selections .~ ((s & #originStream .~ swapped) : ss)+                & #streamHeads .~ (StreamHeadWitness swapped (StreamVersion 1) : filter (\headValue -> headValue.stream /= s.originStream) input.streamHeads),+              input+                & #selections .~ ((s & #acknowledgedLinks .~ V.filter (\l -> l.stream /= target) s.acknowledgedLinks) : ss)+                & #streamHeads .~ filter (\headValue -> headValue.stream /= target) input.streamHeads+            ]+      forM_ variants $ \variant -> assert (compactionManifestDigest (manifest variant) /= compactionManifestDigest m)+    it "round-trips manifests, reports and ledger records through JSON" $ hedgehog $ do+      input <- forAll genInput+      let m = manifest input+          r = reportFor input+          record = CompactionRecord (CompactionId UUID.nil) r (UTCTime (fromGregorian 2026 10 27) 0) "operator"+      Aeson.eitherDecode (Aeson.encode m) === Right m+      Aeson.eitherDecode (Aeson.encode r) === Right r+      Aeson.eitherDecode (Aeson.encode record) === Right record+    it "detects digest corruption on decoding" $ hedgehog $ do+      input <- forAll genInput+      let Aeson.Object object = Aeson.toJSON (manifest input)+          damaged = Aeson.Object (KM.insert "digest" (Aeson.toJSON zeroDigest) object)+      assert (isError (Aeson.fromJSON damaged :: Aeson.Result CompactionManifest))++origin, target, reserved :: StreamName+origin = StreamName "orders-1"+target = StreamName "audit-2026"+reserved = StreamName "$all"++eid :: (Integral a) => a -> EventId+eid n = EventId (UUID.fromWords64 0 (fromIntegral n))++operation :: Text -> CompactionOperation+operation = either (error . show) id . mkCompactionOperation++zeroDigest :: CompactionDigest+zeroDigest = fromJust (mkCompactionDigest (BS.replicate 32 0))++selection :: CompactionSelection+selection = CompactionSelection (EventId (fromJust (UUID.fromText "00000000-0000-7000-8000-0000000000aa"))) origin (StreamVersion 3) (GlobalPosition 42) V.empty++headWitness :: StreamHeadWitness+headWitness = StreamHeadWitness origin (StreamVersion 7)++base :: CompactionManifestInput+base = CompactionManifestInput (StoreIdentity (fromJust (UUID.fromText "00000000-0000-7000-8000-000000000001"))) (operation "compact") RefuseDeadLetters RefuseCausationDependents [headWitness] [selection] Nothing++manifest :: CompactionManifestInput -> CompactionManifest+manifest = either (error . show) id . mkCompactionManifest++changeSelection :: (CompactionSelection -> CompactionSelection) -> CompactionManifestInput+changeSelection f = base & #selections .~ [f selection]++isError :: Aeson.Result a -> Bool+isError (Aeson.Error _) = True+isError _ = False++reportFor :: CompactionManifestInput -> CompactionReport+reportFor input =+  sealCompactionReport+    ( CompactionReport+        (compactionManifestDigest m)+        (manifestStoreIdentity m)+        (manifestOperation m)+        (manifestDeadLetterPolicy m)+        (manifestCausationPolicy m)+        count+        count+        count+        links+        0+        0+        (V.head sels).globalPosition+        (V.last sels).globalPosition+        (manifestStreamHeads m)+        zeroDigest+    )+  where+    m = manifest input+    sels = manifestSelections m+    count = fromIntegral (V.length sels)+    links = fromIntegral (sum (map (V.length . (\s -> s.acknowledgedLinks)) (V.toList sels)))++genInput :: Gen CompactionManifestInput+genInput = do+  size <- Gen.int (Range.linear 1 15)+  seed <- Gen.word64 (Range.linear 100 100000)+  let names = [StreamName ("ψ-" <> T.pack (show n)) | n <- [1 .. size]]+      sels =+        [ CompactionSelection+            (eid (seed + fromIntegral n))+            name+            (StreamVersion 1)+            (GlobalPosition (fromIntegral n))+            (if n == 1 then V.fromList [LinkWitness target (StreamVersion 1), LinkWitness (StreamName "audit-2027") (StreamVersion 2)] else V.empty)+        | (n, name) <- zip [1 :: Int ..] names+        ]+  pure (base & #selections .~ sels & #streamHeads .~ (StreamHeadWitness target (StreamVersion 1) : StreamHeadWitness (StreamName "audit-2027") (StreamVersion 2) : map (\name -> StreamHeadWitness name (StreamVersion 1)) names))

@@ -0,0 +1,214 @@+{-# LANGUAGE OverloadedRecordDot #-}++module Test.CompactionPreview (spec) where++import Control.Concurrent (myThreadId)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (when)+import Data.Aeson qualified as Aeson+import Data.ByteString.Lazy qualified as LBS+import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Data.UUID qualified as UUID+import Data.Vector qualified as V+import Kiroku.Store hiding (id, target)+import Test.CompactionHelpers+import Test.Helpers+import Test.Hspec++spec :: Spec+spec = describe "compaction preview" $ do+  it "returns a deterministic sealed report and leaves every event unchanged" $ withTestStore $ \store -> do+    events <- seed store origin 5+    input <- inputFor store [events !! 1, events !! 2]+    before <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+    first <- readyReport store (manifest input)+    second <- readyReport store (manifest input)+    second `shouldBe` first+    (first.selectedEvents, first.homeMemberships, first.globalMemberships, first.linkMemberships) `shouldBe` (2, 2, 2, 0)+    (first.lowestGlobalPosition, first.highestGlobalPosition) `shouldBe` (GlobalPosition 2, GlobalPosition 3)+    first.reportDigest `shouldBe` compactionReportDigest first+    runStoreIO store (readAllForward (GlobalPosition 0) 100) `shouldReturn` Right before+    runStoreIO store (runTransaction (previewCompactionTx (manifest input))) `shouldReturn` Right (Right (CompactionPreviewReady first))++  it "previews the stream head and logically truncated events" $ withTestStore $ \store -> do+    events <- seed store origin 3+    input <- inputFor store [head events, last events]+    _ <- storeRead =<< runStoreIO store (setStreamTruncateBefore origin (StreamVersion 3))+    report <- readyReport store (manifest input)+    report.selectedEvents `shouldBe` 2++  it "checks acknowledged links per fact, including links of links" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    _ <- storeRead =<< runStoreIO store (linkToStream target [event.eventId])+    _ <- storeRead =<< runStoreIO store (linkToStream mirror [event.eventId])+    input <- inputFor store [event]+    report <- readyReport store (manifest input)+    report.linkMemberships `shouldBe` 2+    let [selection] = input.selections+        unacknowledged = manifest (input & #selections .~ [selection & #acknowledgedLinks .~ V.singleton (LinkWitness target (StreamVersion 1))] & #streamHeads .~ filter (\h -> h.stream /= mirror) input.streamHeads)+    assertRefusals store unacknowledged [CompactionUnexpectedLink event.eventId (LinkWitness mirror (StreamVersion 1))]+    let badLink = LinkWitness target (StreamVersion 2)+        bad = manifest (input & #selections .~ [selection & #acknowledgedLinks .~ V.fromList [badLink, LinkWitness mirror (StreamVersion 1)]])+    assertRefusals store bad [CompactionUnexpectedLink event.eventId (LinkWitness target (StreamVersion 1)), CompactionAcknowledgedLinkMissing event.eventId badLink]++  it "reports missing streams without fabricating event or link refusals" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let [selection] = input.selections+        missing = StreamName "missing"+        changed =+          input+            & #selections .~ [selection & #originStream .~ missing]+            & #streamHeads .~ [StreamHeadWitness missing (StreamVersion 1)]+    assertRefusals store (manifest changed) [CompactionStreamMissing missing]+    let linked =+          input+            & #selections .~ [selection & #acknowledgedLinks .~ V.singleton (LinkWitness missing (StreamVersion 1))]+            & #streamHeads .~ (StreamHeadWitness missing (StreamVersion 1) : input.streamHeads)+    assertRefusals store (manifest linked) [CompactionStreamMissing missing]++  it "reports soft deletion and head drift" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let drifted = manifest (input & #streamHeads .~ [StreamHeadWitness origin (StreamVersion 2)])+    assertRefusals store drifted [CompactionStreamHeadDrift origin (StreamVersion 2) (StreamVersion 1)]+    _ <- storeRead =<< runStoreIO store (softDeleteStream origin)+    assertRefusals store (manifest input) [CompactionStreamSoftDeleted origin]++  it "reports wrong origin versions and global positions, and unknown selected IDs" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let [selection] = input.selections+        actual = WitnessMismatch origin (StreamVersion 1) (Just (GlobalPosition 1))+    assertRefusals store (manifest (input & #selections .~ [selection & #originVersion .~ StreamVersion 2])) [CompactionWitnessMismatch event.eventId actual]+    assertRefusals store (manifest (input & #selections .~ [selection & #globalPosition .~ GlobalPosition 2])) [CompactionWitnessMismatch event.eventId actual]+    assertRefusals store (manifest (input & #selections .~ [selection & #eventId .~ EventId UUID.nil])) [CompactionSelectedEventMissing (EventId UUID.nil)]++  it "detects missing home/global rows and inconsistent original-version columns" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    raw store "BEGIN; ALTER TABLE kiroku.stream_events DISABLE TRIGGER no_update_stream_events; UPDATE kiroku.stream_events SET original_stream_version=2 WHERE stream_id=0; ALTER TABLE kiroku.stream_events ENABLE TRIGGER no_update_stream_events; COMMIT"+    assertRefusals store (manifest input) [CompactionWitnessMismatch event.eventId (WitnessMismatch origin (StreamVersion 2) (Just (GlobalPosition 1)))]+    raw store "BEGIN; SET LOCAL kiroku.enable_hard_deletes='on'; DELETE FROM kiroku.stream_events WHERE stream_id=0; COMMIT"+    assertRefusals store (manifest input) [CompactionWitnessMismatch event.eventId (WitnessMismatch origin (StreamVersion 1) Nothing)]+    -- Restore just the global row through SQL, then remove the home row.+    let EventId uid = event.eventId+        StreamId sid = event.originalStreamId+    raw store ("BEGIN; SET LOCAL kiroku.enable_hard_deletes='on'; INSERT INTO kiroku.stream_events(event_id,stream_id,stream_version,original_stream_id,original_stream_version,category) VALUES ('" <> UUID.toText uid <> "',0,1," <> T.pack (show sid) <> ",1,'orders'); DELETE FROM kiroku.stream_events WHERE stream_id <> 0; COMMIT")+    assertRefusals store (manifest input) [CompactionWitnessMismatch event.eventId (WitnessMismatch origin (StreamVersion 1) (Just (GlobalPosition 1)))]++  it "refuses or counts dead letters according to the sealed policy" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    insertDeadLetterForEvent store "projection" event+    assertRefusals store (manifest input) [CompactionDeadLettersPresent event.eventId 1]+    report <- readyReport store (manifest (input & #deadLetterPolicy .~ RemoveDeadLetters))+    report.deadLettersRemoved `shouldBe` 1++  it "excludes the entire selected set from causation dependents" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    let EventId uid = event.eventId+    _ <- storeRead =<< runStoreIO store (appendToStream target NoStream [makeEvent "Dependent" (Aeson.object []) & #causationId .~ Just uid])+    input <- inputFor store [event]+    assertRefusals store (manifest input) [CompactionCausationDependentsPresent event.eventId 1]+    report <- readyReport store (manifest (input & #causationPolicy .~ AllowDanglingCausation))+    report.causationDependents `shouldBe` 1+    events <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100)+    together <- inputFor store (V.toList events)+    both <- readyReport store (manifest together)+    both.causationDependents `shouldBe` 0++  it "short-circuits a wrong store identity before all other findings" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let wrong = StoreIdentity UUID.nil+        changed = input & #storeIdentity .~ wrong & #streamHeads .~ [StreamHeadWitness origin (StreamVersion 99)]+    assertRefusals store (manifest changed) [CompactionStoreIdentityMismatch wrong input.storeIdentity]++  it "reports active leases and succeeds after release" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let Right owner = mkHistoryRetentionLeaseOwner "preview"+        Right reason = mkHistoryRetentionLeaseReason "test"+        Right duration = mkHistoryRetentionLeaseDuration 60+    lease <- storeRead =<< runStoreIO store (acquireHistoryRetentionLease (HistoryRetentionLeaseRequest owner reason duration))+    Right (Left (CompactionHistoryRetentionActive conflict :| [])) <- runStoreIO store (previewCompaction (manifest input))+    conflict.activeLeaseCount `shouldBe` 1+    _ <- storeRead =<< runStoreIO store (releaseHistoryRetentionLease (HistoryRetentionLeaseHandle (lease ^. #leaseId) owner))+    _ <- readyReport store (manifest input)+    pure ()++  it "returns every independent finding in deterministic order" $ withTestStore $ \store -> do+    [event] <- seed store origin 1+    input <- inputFor store [event]+    let [selection] = input.selections+        missing = StreamName "zz-missing"+        extra = CompactionSelection (EventId UUID.nil) missing (StreamVersion 1) (GlobalPosition 2) V.empty+        changed =+          input+            & #selections .~ [selection & #originVersion .~ StreamVersion 2, extra]+            & #streamHeads .~ [StreamHeadWitness origin (StreamVersion 2), StreamHeadWitness missing (StreamVersion 1)]+    assertRefusals+      store+      (manifest changed)+      [ CompactionStreamHeadDrift origin (StreamVersion 2) (StreamVersion 1),+        CompactionStreamMissing missing,+        CompactionWitnessMismatch event.eventId (WitnessMismatch origin (StreamVersion 1) (Just (GlobalPosition 1)))+      ]++  it "recognizes ledger conflicts and returns stored records without resealing them" $ withTestStore $ \store -> do+    events <- seed store origin 2+    input <- inputFor store events+    let m = manifest input+        StoreIdentity identity = input.storeIdentity+        digest = compactionDigestHex (compactionManifestDigest m)+        headsJson = TE.decodeUtf8 (LBS.toStrict (Aeson.encode input.streamHeads))+    raw store ("INSERT INTO kiroku.event_compactions(manifest_digest,report_digest,store_id,operation,dead_letter_policy,causation_policy,selected_events,home_memberships,global_memberships,link_memberships,dead_letters_removed,causation_dependents,lowest_global_position,highest_global_position,affected_streams) VALUES (decode('" <> digest <> "','hex'),decode(repeat('00',32),'hex'),'" <> UUID.toText identity <> "','stored-operation','refuse','refuse',2,2,2,0,0,0,1,2,'" <> headsJson <> "'::jsonb)")+    Right (Left (CompactionLedgerConflict cid 2 :| [])) <- runStoreIO store (previewCompaction m)+    raw store "BEGIN; SET LOCAL kiroku.enable_hard_deletes='on'; DELETE FROM kiroku.stream_events; DELETE FROM kiroku.events; COMMIT"+    Right (Right (CompactionPreviewAlreadyApplied record)) <- runStoreIO store (previewCompaction m)+    record.compactionId `shouldBe` cid+    record.report.operation `shouldBe` either (error . show) id (mkCompactionOperation "stored-operation")+    compactionDigestHex record.report.reportDigest `shouldBe` T.replicate 64 "0"++  it "uses one caller checkout and emits one typed outcome event" $ do+    ownerThread <- myThreadId+    checkouts <- newIORef (0 :: Int)+    observed <- newIORef []+    let tweak settings =+          settings+            { eventHandler = Just (\event -> modifyIORef' observed (event :)),+              observationHandler = Just $ \case+                ConnectionObservation _ InUseConnectionStatus -> do+                  current <- myThreadId+                  when (current == ownerThread) (modifyIORef' checkouts (+ 1))+                _ -> pure ()+            }+    withTestStoreSettings tweak $ \store -> do+      events <- seed store origin 2+      input <- inputFor store events+      writeIORef observed []+      before <- readIORef checkouts+      _ <- readyReport store (manifest input)+      readIORef checkouts `shouldReturn` (before + 1)+      recorded <- readIORef observed+      [(digest, count)] <- pure [(d, n) | KirokuEventCompactionPreviewed d n <- recorded]+      (digest, count) `shouldBe` (compactionManifestDigest (manifest input), 2)+      let drifted = manifest (input & #streamHeads .~ [StreamHeadWitness origin (StreamVersion 3)])+      _ <- storeRead =<< runStoreIO store (previewCompaction drifted)+      refused <- readIORef observed+      [total] <- pure [n | KirokuEventCompactionRefused _ _ n <- refused]+      total `shouldBe` 1++origin, target, mirror :: StreamName+origin = StreamName "orders-1"+target = StreamName "audit-1"+mirror = StreamName "audit-2"++assertRefusals :: KirokuStore -> CompactionManifest -> [CompactionRefusal] -> Expectation+assertRefusals store m expected = runStoreIO store (previewCompaction m) `shouldReturn` Right (Left (NE.fromList expected))

@@ -0,0 +1,35 @@+module Test.CompactionPreviewMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.List.NonEmpty (NonEmpty)+import Data.UUID qualified as UUID+import Data.Vector qualified as V+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store hiding (id)+import Test.Hspec++spec :: Spec+spec = describe "compaction preview mock" $+  it "dispatches the sealed manifest exactly once and returns the configured result" $ do+    calls <- newIORef (0 :: Int)+    let Right op = mkCompactionOperation "mock"+        identity = StoreIdentity UUID.nil+        heads = [StreamHeadWitness (StreamName "mock-1") (StreamVersion 1)]+        selection = CompactionSelection (EventId UUID.nil) (StreamName "mock-1") (StreamVersion 1) (GlobalPosition 1) V.empty+        Right m = mkCompactionManifest (CompactionManifestInput identity op RefuseDeadLetters RefuseCausationDependents heads [selection] Nothing)+        digest = compactionManifestDigest m+        report = sealCompactionReport (CompactionReport digest identity op RefuseDeadLetters RefuseCausationDependents 1 1 1 0 0 0 (GlobalPosition 1) (GlobalPosition 1) (V.fromList heads) digest)+        result = Right (CompactionPreviewReady report)+    actual <- runEff $ runPreviewMock calls m result (previewCompaction m)+    actual `shouldBe` result+    readIORef calls `shouldReturn` 1++runPreviewMock :: (IOE :> es) => IORef Int -> CompactionManifest -> Either (NonEmpty CompactionRefusal) CompactionPreview -> Eff (Store : es) a -> Eff es a+runPreviewMock calls expected result = interpret_ $ \case+  PreviewCompaction m -> do+    liftIO $ m `shouldBe` expected+    liftIO $ modifyIORef' calls (+ 1)+    pure result+  _ -> error "unexpected Store operation in preview mock"

@@ -0,0 +1,116 @@+module Test.EventReferences (spec) where++import Control.Concurrent (myThreadId)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (forM_, when)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.Map.Strict qualified as Map+import Data.UUID qualified as UUID+import Data.Vector qualified as V+import Kiroku.Store+import Kiroku.Store.Effect (maxEventReferenceBatch)+import Test.Helpers+import Test.Hspec++spec :: Spec+spec = describe "event references" $ do+  it "reports home and global memberships with actual global positions" $ withTestStore $ \store -> do+    event <- firstEvent store+    let eid = event ^. #eventId+    Right refs <- runStoreIO store (lookupEventReferences [eid])+    refs+      `shouldBe` Map.singleton+        eid+        ( EventReferenceInventory+            (V.fromList [GlobalMembership (GlobalPosition 1), HomeMembership (StreamName "order-1") (event ^. #originalStreamId) (StreamVersion 1)])+            0+            0+        )++  it "reports all link targets, including a link of a link and soft-deleted targets" $ withTestStore $ \store -> do+    event <- firstEvent store+    let eid = event ^. #eventId+    forM_ [(StreamName "hub"), (StreamName "mirror")] $ \name -> do+      Right _ <- runStoreIO store (linkToStream name [eid])+      pure ()+    Right _ <- runStoreIO store (softDeleteStream (StreamName "hub"))+    Right refs <- runStoreIO store (lookupEventReferences [eid])+    let ms = (refs Map.! eid) ^. #memberships+    V.length ms `shouldBe` 4+    [name | LinkMembership name _ _ <- V.toList ms] `shouldBe` [(StreamName "hub"), (StreamName "mirror")]+    [name | HomeMembership name _ _ <- V.toList ms] `shouldBe` [(StreamName "order-1")]++  it "counts dead letters across subscriptions" $ withTestStore $ \store -> do+    event <- firstEvent store+    insertDeadLetterForEvent store "proj-a" event+    insertDeadLetterForEvent store "proj-b" event+    Right refs <- runStoreIO store (lookupEventReferences [event ^. #eventId])+    (refs Map.! (event ^. #eventId)) ^. #deadLetterCount `shouldBe` 2++  it "excludes requested causation dependents and deduplicates input ids" $ withTestStore $ \store -> do+    event <- firstEvent store+    let eid@(EventId uid) = event ^. #eventId+        dependent = makeEvent "Dependent" (Aeson.object []) & #causationId .~ Just uid+    Right _ <- runStoreIO store (appendToStream (StreamName "caused-1") NoStream [dependent, dependent])+    Right caused <- runStoreIO store (readStreamForward (StreamName "caused-1") (StreamVersion 0) 10)+    Right single <- runStoreIO store (lookupEventReferences [eid])+    (single Map.! eid) ^. #causationDependentCount `shouldBe` 2+    let ids = [eid, caused V.! 0 ^. #eventId]+    Right refs <- runStoreIO store (lookupEventReferences (ids <> ids))+    (refs Map.! eid) ^. #causationDependentCount `shouldBe` 1+    V.length ((refs Map.! eid) ^. #memberships) `shouldBe` 2+    runStoreIO store (runTransaction (lookupEventReferencesTx ids)) `shouldReturn` Right refs++  it "omits unknown ids and agrees with the transaction combinator" $ withTestStore $ \store -> do+    event <- firstEvent store+    let ids = [event ^. #eventId, EventId UUID.nil]+    Right refs <- runStoreIO store (lookupEventReferences ids)+    Map.keys refs `shouldBe` [event ^. #eventId]+    runStoreIO store (runTransaction (lookupEventReferencesTx ids)) `shouldReturn` Right refs++  it "uses no checkout for empty input and one checkout across batches" $ do+    owner <- myThreadId+    checkouts <- newIORef (0 :: Int)+    let tweak settings =+          settings+            { observationHandler = Just $ \case+                ConnectionObservation _ InUseConnectionStatus -> do+                  current <- myThreadId+                  when (current == owner) (modifyIORef' checkouts (+ 1))+                _ -> pure ()+            }+    withTestStoreSettings tweak $ \store -> do+      beforeEmpty <- readIORef checkouts+      runStoreIO store (lookupEventReferences []) `shouldReturn` Right Map.empty+      readIORef checkouts `shouldReturn` beforeEmpty+      -- Last requested event depends on the first: the exclusion must span batches.+      let firstId = UUID.fromWords64 0 1+          events =+            [ makeEvent "Batch" (Aeson.object [])+                & #eventId .~ Just (EventId (UUID.fromWords64 0 (fromIntegral n)))+                & #causationId .~ (if n == maxEventReferenceBatch + 1 then Just firstId else Nothing)+            | n <- [1 .. maxEventReferenceBatch + 1]+            ]+      forM_ (chunks 2000 events) $ \batch -> do+        Right _ <- runStoreIO store (appendToStream (StreamName "batch-1") AnyVersion batch)+        pure ()+      let ids = [EventId (UUID.fromWords64 0 (fromIntegral n)) | n <- [1 .. maxEventReferenceBatch + 1]]+      beforeRead <- readIORef checkouts+      Right refs <- runStoreIO store (lookupEventReferences ids)+      readIORef checkouts `shouldReturn` (beforeRead + 1)+      Map.size refs `shouldBe` maxEventReferenceBatch + 1+      forM_ (Map.elems refs) $ \inventory -> do+        V.length (inventory ^. #memberships) `shouldBe` 2+        inventory ^. #causationDependentCount `shouldBe` 0++firstEvent :: KirokuStore -> IO RecordedEvent+firstEvent store = do+  Right _ <- runStoreIO store (appendToStream (StreamName "order-1") NoStream [makeEvent "Placed" (Aeson.object [])])+  Right events <- runStoreIO store (readAllForward (GlobalPosition 0) 10)+  pure (V.head events)++chunks :: Int -> [a] -> [[a]]+chunks _ [] = []+chunks size values = let (batch, rest) = splitAt size values in batch : chunks size rest

@@ -0,0 +1,35 @@+module Test.EventReferencesMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.UUID qualified as UUID+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store.Effect (Store (..))+import Kiroku.Store.Read (lookupEventReferences)+import Kiroku.Store.Types+import Test.Hspec++spec :: Spec+spec = describe "event references mock" $+  it "returns the configured identity through one Store effect call" $ do+    calls <- newIORef (0 :: Int)+    let expected = Map.singleton (EventId UUID.nil) (EventReferenceInventory mempty 2 3)+    actual <- runEff $ runIdentityMock calls expected (lookupEventReferences [EventId UUID.nil])+    actual `shouldBe` expected+    readIORef calls `shouldReturn` 1++runIdentityMock ::+  (IOE :> es) =>+  IORef Int ->+  Map EventId EventReferenceInventory ->+  Eff (Store : es) a ->+  Eff es a+runIdentityMock calls expected = interpret_ $ \case+  LookupEventReferences ids -> do+    liftIO $ ids `shouldBe` [EventId UUID.nil]+    liftIO $ modifyIORef' calls (+ 1)+    pure expected+  _ -> error "unexpected Store operation in identity mock"

@@ -11,11 +11,13 @@ import Data.Generics.Labels () import Data.Text qualified as T import Data.Text.Encoding qualified as T+import Data.Vector qualified as V import Hasql.Connection qualified as Connection import Hasql.Connection.Settings qualified as Conn import Hasql.Notifications qualified as Notifications import Kiroku.Store import Kiroku.Test.Postgres (withMigratedTestDatabase)+import Test.CompactionHelpers qualified as CompactionHelpers import Test.Helpers (makeEvent) import Test.Hspec @@ -45,6 +47,10 @@               runStoreIO store $                 appendToStream secondStream NoStream [makeEvent "NotifyGuardSecond" (Aeson.object [])]             waitForPayload payloads (expectedPayload secondStream second)++            Right allEvents <- runStoreIO store (readAllForward (GlobalPosition 0) 10)+            input <- CompactionHelpers.inputFor store [V.head allEvents]+            Right (Right (CompactionAppliedNow _)) <- runStoreIO store (applyCompaction (CompactionHelpers.manifest input))              actual <- atomically (readTVar payloads)             actual

@@ -5,7 +5,7 @@ import Control.Concurrent (myThreadId) import Control.Exception (fromException) import Control.Lens ((^.))-import Control.Monad (forM_, unless)+import Control.Monad (forM_, unless, when) import Data.Aeson (Value (..)) import Data.Aeson qualified as Aeson import Data.Aeson.KeyMap qualified as KeyMap@@ -22,11 +22,13 @@ import Hasql.Statement (Statement, unpreparable) import Hasql.Statement qualified as Statement import Kiroku.Store+import Kiroku.Store.Compaction.SQL qualified as CompactionSQL import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.Stream qualified as Buffer import Kiroku.Store.Subscription.Worker (withLoadCheckpointHookForTest) import Kiroku.Test.Fixtures.CategoryScaling (categoryScalingFixtureSql, categoryScalingHead) import Kiroku.Test.Fixtures.StreamHead (streamHeadFixtureSql)+import Test.CompactionHelpers qualified as CompactionHelpers import Test.DeadLetterQueryPlans qualified as DeadLetterQueryPlans import Test.Helpers (makeEvent, validConsumerGroup, waitWithTimeout, withTestStore, withTestStoreSettings) import Test.Hspec@@ -167,6 +169,59 @@         checkoutsAfter <- readIORef checkouts         checkoutsAfter - checkoutsBefore `shouldBe` 0 +    it "applies a manifest with one caller pool checkout" $ do+      owner <- myThreadId+      checkouts <- newIORef (0 :: Int)+      let tweak settings =+            settings+              { observationHandler = Just $ \case+                  ConnectionObservation _ InUseConnectionStatus -> do+                    current <- myThreadId+                    when (current == owner) (modifyIORef' checkouts (+ 1))+                  _ -> pure ()+              }+      withTestStoreSettings tweak $ \store -> do+        events <- CompactionHelpers.seed store (StreamName "apply-checkout") 1+        input <- CompactionHelpers.inputFor store events+        beforeApply <- readIORef checkouts+        Right (Right (CompactionAppliedNow _)) <- runStoreIO store (applyCompaction (CompactionHelpers.manifest input))+        readIORef checkouts `shouldReturn` (beforeApply + 1)++    it "guards compaction UPDATE, DELETE and TRUNCATE while leaving INSERT unguarded" $+      withTestStore $ \store -> do+        Right shape <- Pool.use (store ^. #pool) (Session.statement () compactionTriggerShapeStmt)+        shape `shouldBe` (6, 2, 0)++    it "previews a manifest with one caller pool checkout" $ do+      owner <- myThreadId+      checkouts <- newIORef (0 :: Int)+      let tweak settings =+            settings+              { observationHandler = Just $ \case+                  ConnectionObservation _ InUseConnectionStatus -> do+                    current <- myThreadId+                    when (current == owner) (modifyIORef' checkouts (+ 1))+                  _ -> pure ()+              }+      withTestStoreSettings tweak $ \store -> do+        events <- CompactionHelpers.seed store (StreamName "preview-checkout") 1+        input <- CompactionHelpers.inputFor store events+        beforePreview <- readIORef checkouts+        _ <- CompactionHelpers.readyReport store (CompactionHelpers.manifest input)+        readIORef checkouts `shouldReturn` (beforePreview + 1)++    it "keeps preview SQL free of row locks and writes" $ do+      let statements =+            [ Statement.toSql CompactionSQL.resolveStreamsStmt,+              Statement.toSql CompactionSQL.ledgerRecordByDigestStmt,+              Statement.toSql CompactionSQL.survivingSelectedEventsStmt,+              Statement.toSql SQL.eventMembershipsStmt,+              Statement.toSql SQL.deadLetterCountsStmt,+              Statement.toSql SQL.causationDependentCountsStmt+            ]+      forM_ statements $ \sql -> forM_ ["FOR UPDATE", "FOR SHARE", "DELETE FROM", "INSERT INTO", "UPDATE ", "SET LOCAL"] $ \forbidden ->+        T.toUpper sql `shouldNotSatisfy` T.isInfixOf forbidden+     it "keeps every ordinary statement free of retention coordination" $ do       let ordinarySql =             [ Statement.toSql SQL.appendExpectedVersion,@@ -180,12 +235,47 @@               Statement.toSql SQL.setStreamTruncateBeforeStmt             ]       forM_ ordinarySql $ \sql ->-        T.toLower sql `shouldNotSatisfy` T.isInfixOf "history_retention"+        forM_ ["history_retention", "store_identity", "event_compactions"] $ \name ->+          T.toLower sql `shouldNotSatisfy` T.isInfixOf name  queryPlanSpec :: Spec queryPlanSpec =   describe "production query plans" $     aroundAll withQueryPlanStore $ do+      it "serves the ledger digest lookup through its unique index" $ \store -> do+        plan <- explainProductionStatement store CompactionSQL.ledgerRecordByDigestStmt [("$1", "decode(repeat('00',32),'hex')")]+        expectIndex "event_compactions_manifest_digest_key" plan++      it "serves newest ledger records through ix_event_compactions_applied_at without Sort" $ \store -> do+        plan <- explainProductionStatement store CompactionSQL.compactionLedgerStmt [("$1", "10")]+        expectIndex "ix_event_compactions_applied_at" plan+        expectNoNodeType "Sort" plan++      it "serves a 500-pair junction delete through stream_events_pkey" $ \store -> do+        let streamKeys = "ARRAY[" <> T.intercalate "," (replicate 500 "1") <> "]::bigint[]"+        plan <- explainProductionStatement store CompactionSQL.deleteJunctionRowsStmt [("$1::uuid[]", inventoryUuidArray), ("$2::bigint[]", streamKeys)]+        expectIndex "stream_events_pkey" plan++      it "resolves a 500-name compaction inventory through ix_streams_stream_name" $ \_ ->+        withTestStore $ \store -> do+          seeded <- Pool.use (store ^. #pool) (Session.script "INSERT INTO streams(stream_name,stream_version) SELECT 'compaction-plan-' || n,0 FROM generate_series(1,50000) n; ANALYZE streams")+          seeded `shouldBe` Right ()+          let names = "ARRAY[" <> T.intercalate "," ["'compaction-plan-" <> T.pack (show n) <> "'" | n <- [1 .. 500 :: Int]] <> "]::text[]"+          plan <- explainProductionStatement store CompactionSQL.resolveStreamsStmt [("$1::text[]", names)]+          expectIndex "ix_streams_stream_name" plan++      it "event membership inventory uses stream_events_pkey" $ \store -> do+        plan <- explainProductionStatement store SQL.eventMembershipsStmt [("$1::uuid[]", inventoryUuidArray)]+        expectIndex "stream_events_pkey" plan++      it "dead-letter counts use ix_dead_letters_event_id" $ \store -> do+        plan <- explainProductionStatement store SQL.deadLetterCountsStmt [("$1::uuid[]", inventoryUuidArray)]+        expectIndex "ix_dead_letters_event_id" plan++      it "causation dependent counts use ix_events_causation_id" $ \store -> do+        plan <- explainProductionStatement store SQL.causationDependentCountsStmt [("$1::uuid[]", inventoryUuidArray)]+        expectIndex "ix_events_causation_id" plan+       it "visible global head lookup uses ux_stream_events_stream_version without Sort" $ \store -> do         plan <-           explainProductionStatement@@ -448,6 +538,15 @@          CASE WHEN n <= 9900 THEN clock_timestamp() ELSE NULL END   FROM generate_series(1, 10000) AS n;   ANALYZE history_retention_leases;+  INSERT INTO event_compactions(manifest_digest,report_digest,store_id,operation,+    dead_letter_policy,causation_policy,selected_events,home_memberships,global_memberships,+    link_memberships,dead_letters_removed,causation_dependents,lowest_global_position,+    highest_global_position,affected_streams,applied_at)+  SELECT decode(lpad(to_hex(n),64,'0'),'hex'),decode(repeat('01',32),'hex'),uuidv7(),+    'structural-fixture','refuse','refuse',1,1,1,0,0,0,1,1,'[]'::jsonb,+    clock_timestamp() - n * interval '1 second'+  FROM generate_series(1,500) n;+  ANALYZE event_compactions;   """  explainProductionStatement ::@@ -646,3 +745,16 @@ planNumber key fields = case KeyMap.lookup key fields of   Just (Number n) -> realToFrac n   _ -> 0++-- Hundreds of distinct IDs exercise the actual maintenance-array plan shape.+inventoryUuidArray :: Text+inventoryUuidArray = "ARRAY[" <> T.intercalate "," ["'00000000-0000-0000-0000-" <> T.justifyRight 12 '0' (T.pack (show n)) <> "'" | n <- [1 .. 500 :: Int]] <> "]::uuid[]"++compactionTriggerShapeStmt :: Statement () (Int64, Int64, Int64)+compactionTriggerShapeStmt =+  unpreparable+    "SELECT count(*),count(*) FILTER (WHERE (tgtype & 16) <> 0),count(*) FILTER (WHERE (tgtype & 4) <> 0) FROM pg_trigger WHERE NOT tgisinternal AND tgrelid IN ('kiroku.store_identity'::regclass,'kiroku.event_compactions'::regclass)"+    E.noParams+    (D.singleRow ((,,) <$> col D.int8 <*> col D.int8 <*> col D.int8))+  where+    col = D.column . D.nonNullable

@@ -0,0 +1,43 @@+module Test.StoreIdentity (spec) where++import Control.Lens ((^.))+import Data.Either (isLeft)+import Data.Generics.Labels ()+import Data.Text (Text)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Kiroku.Store+import Test.Helpers (truncateRejected, withTestStore)+import Test.Hspec++spec :: Spec+spec = describe "store identity" $ do+  it "is stable across calls, pooled connections and transaction reads" $+    withTestStore $ \store -> do+      Right first <- runStoreIO store storeIdentity+      runStoreIO store storeIdentity `shouldReturn` Right first+      runStoreIO store (runTransaction storeIdentityTx) `shouldReturn` Right first++  it "is shared by separate databases cloned from the migrated template" $+    withTestStore $ \first -> withTestStore $ \second -> do+      Right identity <- runStoreIO first storeIdentity+      runStoreIO second storeIdentity `shouldReturn` Right identity++  it "rejects identity and ledger mutation without the hard-delete guard" $+    withTestStore $ \store -> do+      rejects store "UPDATE store_identity SET store_id = gen_random_uuid()"+      rejects store "DELETE FROM store_identity"+      truncateRejected store "store_identity" `shouldReturn` True+      seeded <-+        Pool.use (store ^. #pool) $+          Session.script+            "INSERT INTO event_compactions (manifest_digest, report_digest, store_id, operation, dead_letter_policy, causation_policy, selected_events, home_memberships, global_memberships, link_memberships, dead_letters_removed, causation_dependents, lowest_global_position, highest_global_position, affected_streams) VALUES (decode(repeat('ab',32),'hex'), decode(repeat('cd',32),'hex'), gen_random_uuid(), 'trigger-test', 'refuse', 'refuse', 1, 1, 1, 0, 0, 0, 1, 1, '[]'::jsonb)"+      seeded `shouldBe` Right ()+      rejects store "UPDATE event_compactions SET operation = 'changed'"+      rejects store "DELETE FROM event_compactions"+      truncateRejected store "event_compactions" `shouldReturn` True++rejects :: KirokuStore -> Text -> Expectation+rejects store sql = do+  result <- Pool.use (store ^. #pool) (Session.script sql)+  result `shouldSatisfy` isLeft

@@ -0,0 +1,32 @@+module Test.StoreIdentityMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.UUID qualified as UUID+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store.Effect (Store (..))+import Kiroku.Store.Read (storeIdentity)+import Kiroku.Store.Types (StoreIdentity (..))+import Test.Hspec++spec :: Spec+spec = describe "store identity mock" $+  it "returns the configured identity through one Store effect call" $ do+    calls <- newIORef (0 :: Int)+    let expected = StoreIdentity UUID.nil+    actual <- runEff $ runIdentityMock calls expected storeIdentity+    actual `shouldBe` expected+    readIORef calls `shouldReturn` 1++runIdentityMock ::+  (IOE :> es) =>+  IORef Int ->+  StoreIdentity ->+  Eff (Store : es) a ->+  Eff es a+runIdentityMock calls expected = interpret_ $ \case+  GetStoreIdentity -> do+    liftIO $ modifyIORef' calls (+ 1)+    pure expected+  _ -> error "unexpected Store operation in identity mock"

@@ -35,7 +35,7 @@       events <- seedEvents store 7       mapM_ (insertDeadLetterForEvent store "page") (take 5 events)       mapM_ (insertDeadLetterForEvent store "other") (drop 5 events)-      let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "page")) {limit = validLimit 2}+      let query = defaultSubscriptionDeadLetterQuery (SubscriptionName "page") & #limit .~ validLimit 2       first <- page store query       second <- page store query {after = first ^. #nextCursor}       third <- page store query {after = second ^. #nextCursor}@@ -50,7 +50,7 @@     it "merges historical members and resolves same-position ties with an exclusive cursor" $ \store -> do       [event] <- seedEvents store 1       forM_ [0, 1, 9] $ \member -> insertDeadLetterWith store "ties" member (object []) "tie" 2 event-      let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "ties")) {limit = validLimit 1}+      let query = defaultSubscriptionDeadLetterQuery (SubscriptionName "ties") & #limit .~ validLimit 1       first <- page store query       second <- page store query {after = first ^. #nextCursor}       third <- page store query {after = second ^. #nextCursor}@@ -79,7 +79,7 @@     it "keeps a cursor valid after its event and row are hard-deleted" $ \store -> do       events <- seedEvents store 3       mapM_ (insertDeadLetterForEvent store "delete") events-      let query = (defaultSubscriptionDeadLetterQuery (SubscriptionName "delete")) {limit = validLimit 1}+      let query = defaultSubscriptionDeadLetterQuery (SubscriptionName "delete") & #limit .~ validLimit 1       first <- page store query       Right _ <- runStoreIO store $ hardDeleteStream (StreamName "dead-3")       older <- page store query {after = first ^. #nextCursor, limit = validLimit 10}