keiro-0.12.0.0: test/VersionedRebuildSpec.hs
{-# LANGUAGE MultilineStrings #-}
module VersionedRebuildSpec
( spec,
)
where
import CatalogSpec
( asyncProjectionId,
auditTargetId,
bridgeCatalog,
bridgeRevisionV1,
bridgeRevisionV2,
catalogAsyncProjection,
counterBinding,
counterReadContract,
counterTargetId,
inlineProjectionId,
mainGroupId,
)
import Contravariant.Extras (contrazip2, contrazip3)
import Control.Concurrent (forkIO, threadDelay)
import Control.Concurrent.MVar (isEmptyMVar, newEmptyMVar, putMVar, takeMVar)
import Control.Exception (bracket)
import Data.Aeson qualified as Aeson
import Data.Aeson.KeyMap qualified as KeyMap
import Data.ByteString (ByteString)
import Data.ByteString qualified as ByteString
import Data.Functor.Contravariant ((>$<))
import Data.Int (Int32)
import Data.List qualified as List
import Data.Map.Strict qualified as Map
import Data.Text qualified as Text
import Data.Text.Encoding qualified as Text.Encoding
import Data.Time (UTCTime (..), diffUTCTime, secondsToDiffTime)
import Data.Time.Calendar (Day (ModifiedJulianDay))
import Data.UUID qualified as UUID
import Data.UUID.V5 qualified as UUID.V5
import Data.Vector qualified as Vector
import Effectful (Eff, IOE)
import Effectful.Error.Static (Error)
import Hasql.Connection.Settings qualified as ConnectionSettings
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Pool qualified as Pool
import Hasql.Pool.Config qualified as PoolConfig
import Hasql.Session qualified as Session
import Hasql.Statement (Statement, preparable)
import Hasql.Transaction qualified as Tx
import Hasql.Transaction.Sessions qualified as TxSessions
import Keiro.Connection (qualifyTable)
import Keiro.Prelude
import Keiro.Projection (CatalogAsyncApplyOutcome (..), applyAsyncProjectionFromCatalog)
import Keiro.Projection.Catalog
import Keiro.Projection.Catalog qualified as Catalog
import Keiro.Projection.Catalog.Operations qualified as CatalogOperations
import Keiro.ReadModel (ReadModel (..))
import Keiro.ReadModel.External (reconcileExternalReadContracts)
import Keiro.ReadModel.Rebuild
import Keiro.Test.Postgres (Fixture, withFreshDatabase, withFreshStore)
import Kiroku.Store qualified as Store
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Error (StoreError (..))
import Kiroku.Store.HistoryRetention
( HistoryRetentionLeaseRequest (..),
HistoryRetentionRenewalError (..),
mkHistoryRetentionLeaseDuration,
mkHistoryRetentionLeaseOwner,
mkHistoryRetentionLeaseReason,
)
import Kiroku.Store.Transaction qualified as StoreTransaction
import Kiroku.Store.Types
( CategoryName (..),
EventData (..),
EventId (..),
EventType (..),
ExpectedVersion (..),
GlobalPosition (..),
RecordedEvent (..),
StreamId (..),
StreamName (..),
StreamVersion (..),
)
import Test.Hspec
spec :: Fixture -> Spec
spec fixture = do
describe "schema-versioned rebuild lifecycle" $
around (withFreshStore fixture) $ do
it "persists serving and candidate generations and resumes the same run without reprovisioning" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
let request = versionedRequest "versioned-retry" physicalTargets
first <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
second <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
second `shouldBe` first
length (first ^. #candidateGenerations) `shouldBe` 2
map (^. #lifecycle) (first ^. #candidateGenerations)
`shouldBe` [GenerationStaging, GenerationStaging]
runStatement store () activeLifecycleFactsStmt
`shouldReturn` ("rebuilding-versioned", True, True, "counter-v1", 0, 2, 2, 1, 1)
conflict <-
expectStore store (beginVersionedRebuild catalog (request & #cutoverThreshold .~ 99))
conflict
`shouldSatisfy` \case
Left (VersionedRunIdentityConflict _ detail) -> "threshold" `Text.isInfixOf` detail
_ -> False
dedupConflict <-
expectStore store (beginVersionedRebuild catalog (request & #promotionDedupLimit .~ 99))
dedupConflict
`shouldSatisfy` \case
Left (VersionedRunIdentityConflict _ detail) -> "dedup limit" `Text.isInfixOf` detail
_ -> False
it "rolls provisioner failures back with no run, lease, or untracked sibling" $ \store -> do
setupBridge store
(failingCatalog, physicalTargets) <-
validatedBridgeFrom
( replaceCandidateProvisionerInCatalog
counterTargetId
( \provisioner ->
provisioner
{ provisionTarget = \targetContext -> do
createV2Counter targetContext
Tx.sql "SELECT 1 / 0"
}
)
runtimeBridgeCatalog
)
registerBridge store failingCatalog
result <-
Store.runStoreIO
store
(beginVersionedRebuild failingCatalog (versionedRequest "versioned-provision-failure" physicalTargets))
result `shouldSatisfy` isLeft
runStatement store () rolledBackLifecycleFactsStmt
`shouldReturn` ("live", 0, 0, 0, 0)
it "rolls typed schema-validation failure back after candidate DDL" $ \store -> do
setupBridge store
(failingCatalog, physicalTargets) <-
validatedBridgeFrom
( replaceCandidateProvisionerInCatalog
counterTargetId
( \provisioner ->
provisioner
{ validateTarget =
Just
( \_ ->
pure
( Left
[TargetSchemaViolation "shape.counter-v2" "subtotal column was rejected by the application validator"]
)
)
}
)
runtimeBridgeCatalog
)
registerBridge store failingCatalog
result <-
expectStore store (beginVersionedRebuild failingCatalog (versionedRequest "versioned-validation-failure" physicalTargets))
result
`shouldSatisfy` \case
Left (VersionedSchemaValidationFailed targetId violations) ->
targetId == counterTargetId && length violations == 1
_ -> False
runStatement store () rolledBackLifecycleFactsStmt
`shouldReturn` ("live", 0, 0, 0, 0)
it "uses the restricted clone path only for an exact-shape repair" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridgeFrom cloneBridgeCatalog
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-clone" 3
let request =
versionedRequest "versioned-clone" physicalTargets
& #targetMode
.~ RestrictedClone
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
final <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 12
final ^. #phase `shouldBe` VersionedPromoted
final ^. #servingRevisionId `shouldBe` identity mkProjectionRevisionId "counter-v2"
runStatement store () servingCountsStmt `shouldReturn` (3, 3)
runStatement store () cloneShapeStmt `shouldReturn` True
it "remaps cloned primary-key and identity-sequence names during promotion" $ \store -> do
runScript store identityBridgeSql
(catalog, physicalTargets) <- validatedBridgeFrom identityCloneBridgeCatalog
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-identity-clone" 2
let request =
versionedRequest "versioned-identity-clone" physicalTargets
& #targetMode
.~ RestrictedClone
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
final <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 12
final ^. #phase `shouldBe` VersionedPromoted
runStatement store () identityCloneObjectsStmt `shouldReturn` True
it "returns typed restricted-clone findings without leaving a candidate" $ \store -> do
setupBridge store
runScript store cloneTriggerSql
(catalog, physicalTargets) <- validatedBridgeFrom cloneBridgeCatalog
registerBridge store catalog
let request =
versionedRequest "versioned-clone-refused" physicalTargets
& #targetMode
.~ RestrictedClone
refused <- expectStore store (beginVersionedRebuild catalog request)
refused
`shouldSatisfy` \case
Left (VersionedCloneRefused targetId table findings) ->
targetId == counterTargetId
&& table == QualifiedTable "app" "counter"
&& findings == ["triggers"]
_ -> False
runStatement store () rolledBackLifecycleFactsStmt
`shouldReturn` ("live", 0, 0, 0, 0)
it "refuses a deterministic staging-name collision and rolls earlier target work back" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
let runId = run "versioned-name-collision"
collision = candidateTable runId counterTargetId
runScript
store
( Text.Encoding.encodeUtf8
( "CREATE TABLE "
<> qualifyTable (collision ^. #schemaName) (collision ^. #tableName)
<> " (sentinel bigint NOT NULL)"
)
)
result <-
expectStore store (beginVersionedRebuild catalog (rebuildRequestFor runId physicalTargets))
collisionOid <- relationOidFor store collision
result
`shouldBe` Left (VersionedStagingNameCollision counterTargetId collision collisionOid)
runStatement store () collisionRollbackFactsStmt
`shouldReturn` ("live", 0, 0, 0, 1)
it "abandons staging generations, releases retention, and repeats as a no-op" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
let request = versionedRequest "versioned-abandon" physicalTargets
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
first <- expectStore store (abandonVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
second <- expectStore store (abandonVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
first ^. #alreadyAbandoned `shouldBe` False
second ^. #alreadyAbandoned `shouldBe` True
map (^. #lifecycle) (first ^. #droppedGenerations)
`shouldBe` replicate 2 GenerationDropped
for_ (handle ^. #candidateGenerations) $ \generation ->
runStatement store (generation ^. #physicalTable) relationExistsStmt `shouldReturn` False
runStatement store () abandonedLifecycleFactsStmt
`shouldReturn` ("serving-versioned", True, True, "abandoned", 2, 1, 2)
it "dispatches async writes through the persisted serving revision and fails closed when code is absent" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
let request = versionedRequest "versioned-dispatch" physicalTargets
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
legacyView <- expectStore store (lookupProjectionRebuildGroup mainGroupId)
legacyView ^? _Just . #status
`shouldBe` Just (UnknownGroupStatus "rebuilding-versioned")
replayed <-
expectStore
store
(applyVersionedReplayEvent catalog (request ^. #rebuildRunId) (recorded 99))
replayed `shouldBe` Right 1
verified <-
expectStore
store
(verifyVersionedCandidate catalog (request ^. #rebuildRunId))
verified `shouldBe` Right ()
first <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog catalog asyncProjectionId catalogAsyncProjection (recorded 1))
)
first `shouldBe` CatalogAsyncApplied
runStatement store () servingCountsStmt `shouldReturn` (0, 1)
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` 1
runScript store promoteDispatchMetadataSql
second <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog catalog asyncProjectionId catalogAsyncProjection (recorded 2))
)
second `shouldBe` CatalogAsyncApplied
runStatement store () servingCountsStmt `shouldReturn` (0, 1)
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable)
`shouldReturn` if generation ^. #targetId == counterTargetId then 1 else 2
(v1Only, _) <- validatedBridgeFrom runtimeV1OnlyCatalog
missing <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog v1Only asyncProjectionId catalogAsyncProjection (recorded 3))
)
missing
`shouldBe` CatalogAsyncServingRevisionUnavailable
mainGroupId
(identity mkProjectionRevisionId "counter-v2")
runStatement store () dispatchDedupCountStmt `shouldReturn` 2
runStatement store () servingCountsStmt `shouldReturn` (0, 1)
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable)
`shouldReturn` if generation ^. #targetId == counterTargetId then 1 else 2
it "captures a durable final head and atomically promotes every target" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request = versionedRequest "versioned-promote" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
rebuildingStatus <- expectStore store (lookupProjectionGroupStatus mainGroupId)
rebuildingStatus ^? _Just . #lifecyclePhase
`shouldBe` Just "rebuilding-versioned"
rebuildingStatus ^? _Just . #readsAllowed `shouldBe` Just True
rebuildingStatus ^? _Just . #writesAllowed `shouldBe` Just True
rebuildingStatus ^? _Just . #servingRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v1"))
rebuildingStatus ^? _Just . #servingEpoch `shouldBe` Just 0
rebuildingStatus ^? _Just . #activeRunId
`shouldBe` Just (Just (request ^. #rebuildRunId))
rebuildingStatus ^? _Just . #candidateRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v2"))
rebuildingStatus ^? _Just . #candidateRebuildPosition
`shouldBe` Just (Just (GlobalPosition 0))
rebuildingStatus ^? _Just . #candidateRebuildHead
`shouldBe` Just (Just (GlobalPosition 0))
final <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
final ^. #phase `shouldBe` VersionedPromoted
final ^. #servingRevisionId `shouldBe` identity mkProjectionRevisionId "counter-v2"
final ^. #servingEpoch `shouldBe` 1
map (^. #lifecycle) (final ^. #servingGenerations)
`shouldBe` [GenerationServing, GenerationServing]
runStatement store () promotedLifecycleFactsStmt
`shouldReturn` ("serving-versioned", True, True, "counter-v2", 1, "promoted", 2, 2, 1)
runStatement store () promotedCounterShapeStmt `shouldReturn` True
promotedStatus <- expectStore store (lookupProjectionGroupStatus mainGroupId)
promotedStatus ^? _Just . #lifecyclePhase
`shouldBe` Just "serving-versioned"
promotedStatus ^? _Just . #servingRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v2"))
promotedStatus ^? _Just . #servingEpoch `shouldBe` Just 1
promotedStatus ^? _Just . #activeRunId `shouldBe` Just Nothing
promotedStatus ^? _Just . #candidateRevisionId `shouldBe` Just Nothing
promotedStatus ^? _Just . #candidateRebuildPosition `shouldBe` Just Nothing
promotedStatus ^? _Just . #candidateRebuildHead `shouldBe` Just Nothing
promotedStatus ^? _Just . #lastPromotedAt `shouldSatisfy` maybe False isJust
it "keeps an additive all-row v1 contract on the old generation during replay and projects the promoted table to its stable result type" $ \store -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom compatibleExternalReadCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-external-compatible" 2
let request = versionedRequest "versioned-external-compatible" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
expectStore store (reconcileExternalReadContracts catalog) >>= requireRight
runStatement store () externalV1RowsStmt `shouldReturn` [(100, 42)]
ready <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
ready ^. #phase `shouldBe` VersionedCutoverReplaying
runStatement store () externalV1RowsStmt `shouldReturn` [(100, 42)]
promoted <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
promoted ^. #phase `shouldBe` VersionedPromoted
promotedRows <- runStatement store () externalV1RowsStmt
promotedRows `shouldSatisfy` (not . null)
promotedRows `shouldSatisfy` (all ((== 10) . snd))
promotedRows `shouldSatisfy` (all ((/= 100) . fst))
it "activates a breaking v2 contract atomically and fails the old contract with KR003" $ \store -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom breakingExternalReadCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-external-breaking" 2
let request = versionedRequest "versioned-external-breaking" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
expectStore store (reconcileExternalReadContracts catalog) >>= requireRight
runStatement store () externalV1RowsStmt `shouldReturn` [(100, 42)]
_ <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
runStatement store () externalV1RowsStmt `shouldReturn` [(100, 42)]
promoted <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
promoted ^. #phase `shouldBe` VersionedPromoted
oldRead <- Store.runStoreIO store (Store.runTransaction (Tx.statement () externalV1RowsStmt))
oldRead `shouldSatisfy` hasSqlState "KR003"
newRows <- runStatement store () externalV2RowsStmt
newRows `shouldSatisfy` (not . null)
newRows `shouldSatisfy` (all (\(_, subtotal, tax, total) -> subtotal == 8 && tax == 2 && total == 10))
it "keeps v1 current after a breaking promotion only through an explicit compatibility implementation" $ \store -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom compatibilityImplementationCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-external-compatibility-implementation" 2
let request = versionedRequest "versioned-external-compatibility-implementation" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
expectStore store (reconcileExternalReadContracts catalog) >>= requireRight
_ <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
compatibleRows <- runStatement store () externalV1RowsStmt
compatibleRows `shouldSatisfy` (not . null)
compatibleRows `shouldSatisfy` (all ((== 10) . snd))
v2Rows <- runStatement store () externalV2RowsStmt
v2Rows `shouldSatisfy` (all (\(_, subtotal, tax, total) -> subtotal == 8 && tax == 2 && total == 10))
it "converges across two replay rounds while live v1 stays serving and backfills async dedup" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-live" 4
originalEvents <-
expectStore store (Store.readCategory (CategoryName "counter") (GlobalPosition 0) 10)
let request =
versionedRequest "versioned-converge" physicalTargets
& #cutoverThreshold
.~ 0
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
first <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
first ^. #phase `shouldBe` VersionedReplayRunning
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` 2
live <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog catalog asyncProjectionId catalogAsyncProjection (recorded 101))
)
live `shouldBe` CatalogAsyncApplied
runStatement store () servingCountsStmt `shouldReturn` (0, 1)
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` 2
appendVersionedEvents store "counter-later" 2
second <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
allVersionedComplete (second ^. #sources) `shouldBe` True
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` 4
extended <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
extended ^. #capturedHead `shouldBe` GlobalPosition 6
allVersionedComplete (extended ^. #sources) `shouldBe` False
replayedAgain <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
allVersionedComplete (replayedAgain ^. #sources) `shouldBe` True
final <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
final ^. #phase `shouldBe` VersionedPromoted
runStatement store () servingCountsStmt `shouldReturn` (6, 6)
runStatement store () convergeDedupCountStmt `shouldReturn` 7
redelivered <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog catalog asyncProjectionId catalogAsyncProjection (Vector.head originalEvents))
)
redelivered `shouldBe` CatalogAsyncDuplicate
runStatement store () servingCountsStmt `shouldReturn` (6, 6)
(v1Only, _) <- validatedBridgeFrom runtimeV1OnlyCatalog
refused <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog v1Only asyncProjectionId catalogAsyncProjection (recorded 102))
)
refused
`shouldBe` CatalogAsyncServingRevisionUnavailable
mainGroupId
(identity mkProjectionRevisionId "counter-v2")
it "refuses oversized staged dedup evidence before fencing and succeeds after checkpoint lag is reduced" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-large-dedup-lag" 20
let request =
versionedRequest "versioned-dedup-limit" physicalTargets
& #replayPageSize
.~ 4
& #cutoverThreshold
.~ 0
& #promotionDedupLimit
.~ 5
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
replayed <- driveVersionedReplayComplete store catalog (request ^. #rebuildRunId) 10
replayed ^. #stagedDedupCount `shouldBe` 20
refused <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
refused
`shouldBe` Left
(VersionedPromotionDedupLimitExceeded (request ^. #rebuildRunId) 5 20)
unfenced <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
unfenced ^. #phase `shouldBe` VersionedReplayRunning
unfenced ^. #dedupProvisionalHead `shouldBe` Nothing
runStatement store () activeLifecycleFactsStmt
`shouldReturn` ("rebuilding-versioned", True, True, "counter-v1", 0, 2, 2, 1, 1)
runStatement store ("catalog-async-subscription", 20) upsertSubscriptionCursorStmt
promoted <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
promoted ^. #phase `shouldBe` VersionedPromoted
promoted ^. #stagedDedupCount `shouldBe` 0
runStatement store () servingCountsStmt `shouldReturn` (20, 20)
it "uses the Kiroku lease to refuse hard deletion for the full active run" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
appendVersionedEvents store "counter-retained" 1
let request = versionedRequest "versioned-retention" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
blocked <- Store.runStoreIO store (Store.hardDeleteStream (StreamName "counter-retained"))
blocked
`shouldSatisfy` \case
Left HistoryRetentionActive {} -> True
_ -> False
_ <- expectStore store (abandonVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
deleted <- Store.runStoreIO store (Store.hardDeleteStream (StreamName "counter-retained"))
deleted `shouldSatisfy` isRight
it "fails closed when the original retention lease expires" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
let request =
versionedRequest "versioned-expired-retention" physicalTargets
& #retentionLeaseRequest
. #duration
.~ requireIdentity (mkHistoryRetentionLeaseDuration (secondsToDiffTime 1))
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
threadDelay 1_200_000
renewal <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
renewal
`shouldSatisfy` \case
Left (VersionedRetentionRenewalFailed runId HistoryRetentionRenewalExpired) ->
runId == request ^. #rebuildRunId
_ -> False
failed <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
failed ^. #phase `shouldBe` VersionedFailed
runStatement store () expiredRetentionFactsStmt
`shouldReturn` ("failed-versioned", True, False, "failed", "retention.renewal-failed")
it "revalidates candidate DDL under the cutover locks and resumes after repair" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridgeFrom raceBridgeCatalog
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request = versionedRequest "versioned-ddl-race" physicalTargets
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
ready <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
ready ^. #phase `shouldBe` VersionedCutoverReplaying
let counterCandidate =
fromMaybe
(error "counter candidate missing")
(List.find ((== counterTargetId) . (^. #targetId)) (handle ^. #candidateGenerations))
runScript
store
( Text.Encoding.encodeUtf8
( "ALTER TABLE "
<> qualifyTable
(counterCandidate ^. #physicalTable . #schemaName)
(counterCandidate ^. #physicalTable . #tableName)
<> " ADD COLUMN rogue bigint"
)
)
raced <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
raced
`shouldSatisfy` \case
Left (VersionedObservedShapeMismatch targetId _ actual) ->
targetId == counterTargetId && "rogue" `Text.isSuffixOf` actual
_ -> False
stillReady <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
stillReady ^. #phase `shouldBe` VersionedCutoverReplaying
runStatement store () servingCountsStmt `shouldReturn` (0, 0)
runScript
store
( Text.Encoding.encodeUtf8
( "ALTER TABLE "
<> qualifyTable
(counterCandidate ^. #physicalTable . #schemaName)
(counterCandidate ^. #physicalTable . #tableName)
<> " DROP COLUMN rogue"
)
)
repaired <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
repaired ^. #phase `shouldBe` VersionedPromoted
it "previews retired blockers and drops only an unreferenced generation" $ \store -> do
setupExternalBridge store
runScript store retiredReaderSql
(catalog, physicalTargets) <- validatedBridgeFrom compatibleExternalReadCatalog
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request = versionedRequest "versioned-retired-drop" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
_ <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
retired <- expectStore store listVersionedRetiredGenerations
length retired `shouldBe` 2
let counterRetired =
fromMaybe
(error "retired counter generation missing")
(List.find ((== counterTargetId) . (^. #targetId)) retired)
contractBlocked <-
expectStore store (previewVersionedRetiredDrop catalog (counterRetired ^. #generationId))
>>= requireRight
contractBlocked ^. #supportedReadContracts `shouldBe` ["counter_reader/v1"]
contractBlocked ^. #droppable `shouldBe` False
noContractCatalog <-
fst <$> validatedBridgeFrom (runtimeBridgeCatalog {externalReadContracts = []})
dependencyBlocked <-
expectStore
store
(previewVersionedRetiredDrop noContractCatalog (counterRetired ^. #generationId))
>>= requireRight
dependencyBlocked ^. #supportedReadContracts `shouldBe` []
dependencyBlocked ^. #externalDependencies `shouldSatisfy` (not . null)
refused <-
expectStore
store
(dropVersionedRetiredGeneration noContractCatalog (counterRetired ^. #generationId))
refused
`shouldSatisfy` \case
Left (VersionedRetiredDropBlocked generationId blockers) ->
generationId == counterRetired ^. #generationId
&& any ("postgres-dependency:" `Text.isPrefixOf`) blockers
_ -> False
runScript store "DROP VIEW app.retired_counter_reader"
clear <-
expectStore
store
(previewVersionedRetiredDrop noContractCatalog (counterRetired ^. #generationId))
>>= requireRight
clear ^. #droppable `shouldBe` True
dropped <-
expectStore
store
(dropVersionedRetiredGeneration noContractCatalog (counterRetired ^. #generationId))
>>= requireRight
dropped ^. #alreadyDropped `shouldBe` False
runStatement store (counterRetired ^. #physicalTable) relationExistsStmt `shouldReturn` False
secondDrop <-
expectStore
store
(dropVersionedRetiredGeneration noContractCatalog (counterRetired ^. #generationId))
>>= requireRight
secondDrop ^. #alreadyDropped `shouldBe` True
describe "schema-versioned cutover concurrency" $
around (withFreshDatabase fixture) $ do
it "bounds writer-fence row contention and remains unfenced" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupBridge store
runScript store servingOnlyRowsSql
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request =
versionedRequest "versioned-writer-fence-timeout" physicalTargets
& #cutoverLockTimeoutMs
.~ 100
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
_ <- driveVersionedReplayComplete store catalog (request ^. #rebuildRunId) 5
holderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
( do
void (Tx.statement (rebuildGroupIdText mainGroupId) lockGroupRowStmt)
Tx.sql "SELECT pg_sleep(1)"
)
)
>>= putMVar holderDone
waitForGroupRowLock pool 50
startedAt <- getCurrentTime
timedOut <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
finishedAt <- getCurrentTime
timedOut
`shouldBe` Right
(Left (VersionedCutoverDeadlineExceeded (request ^. #rebuildRunId) "writer-fence"))
diffUTCTime finishedAt startedAt `shouldSatisfy` (< 0.35)
stillRunning <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
stillRunning ^. #phase `shouldBe` VersionedReplayRunning
runStatement store () servingCountsStmt `shouldReturn` (1, 1)
_ <- expectPoolUsage =<< takeMVar holderDone
promoted <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
promoted ^. #phase `shouldBe` VersionedPromoted
it "bounds promotion group-row contention without undoing prepared evidence" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupBridge store
runScript store servingOnlyRowsSql
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request =
versionedRequest "versioned-promotion-row-timeout" physicalTargets
& #cutoverLockTimeoutMs
.~ 100
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
_ <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
prepared <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
prepared ^. #promotionPrepared `shouldBe` True
safetyBefore <-
(,)
<$> runStatement store () repairDedupCountStmt
<*> runStatement store () repairCheckpointStmt
holderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
( do
void (Tx.statement (rebuildGroupIdText mainGroupId) lockGroupRowStmt)
Tx.sql "SELECT pg_sleep(1)"
)
)
>>= putMVar holderDone
waitForGroupRowLock pool 50
startedAt <- getCurrentTime
timedOut <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
finishedAt <- getCurrentTime
timedOut
`shouldBe` Right
(Left (VersionedCutoverDeadlineExceeded (request ^. #rebuildRunId) "promotion-group"))
diffUTCTime finishedAt startedAt `shouldSatisfy` (< 0.35)
stillPrepared <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
stillPrepared ^. #promotionPrepared `shouldBe` True
safetyAfter <-
(,)
<$> runStatement store () repairDedupCountStmt
<*> runStatement store () repairCheckpointStmt
safetyAfter `shouldBe` safetyBefore
_ <- expectPoolUsage =<< takeMVar holderDone
promoted <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
promoted ^. #phase `shouldBe` VersionedPromoted
it "bounds two independently contended target relations, keeps readers on v1, and resumes promotion" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupBridge store
runScript store servingOnlyRowsSql
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request =
versionedRequest "versioned-lock-timeout" physicalTargets
& #cutoverLockTimeoutMs
.~ 500
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
ready <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
ready ^. #phase `shouldBe` VersionedCutoverReplaying
prepared <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
prepared ^. #promotionPrepared `shouldBe` True
safetyBefore <-
(,)
<$> runStatement store () repairDedupCountStmt
<*> runStatement store () repairCheckpointStmt
counterReaderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
( do
count <- Tx.statement () counterCountStmt
Tx.sql "SELECT pg_sleep(0.3)"
pure count
)
)
>>= putMVar counterReaderDone
waitForCounterReader pool 50
auditReaderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
( do
count <- Tx.statement () auditCountStmt
Tx.sql "SELECT pg_sleep(1)"
pure count
)
)
>>= putMVar auditReaderDone
waitForAuditReader pool 50
startedAt <- getCurrentTime
timedOut <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
finishedAt <- getCurrentTime
timedOut
`shouldBe` Right
(Left (VersionedCutoverDeadlineExceeded (request ^. #rebuildRunId) "target-relations"))
diffUTCTime finishedAt startedAt `shouldSatisfy` (< 0.65)
stillReady <- expectStore store (inspectVersionedRebuild (request ^. #rebuildRunId)) >>= requireRight
stillReady ^. #phase `shouldBe` VersionedCutoverReplaying
stillReady ^. #promotionPrepared `shouldBe` True
safetyAfter <-
(,)
<$> runStatement store () repairDedupCountStmt
<*> runStatement store () repairCheckpointStmt
safetyAfter `shouldBe` safetyBefore
counterReaderCount <- expectPoolUsage =<< takeMVar counterReaderDone
counterReaderCount `shouldBe` 1
auditReaderCount <- expectPoolUsage =<< takeMVar auditReaderDone
auditReaderCount `shouldBe` 1
runStatement store () servingCountsStmt `shouldReturn` (1, 1)
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` 0
promoted <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
promoted ^. #phase `shouldBe` VersionedPromoted
runStatement store () servingCountsStmt `shouldReturn` (0, 0)
it "exposes only the old or new status tuple during a promotion transaction" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridge
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request = versionedRequest "versioned-status-atomicity" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
beforePromotion <- expectStore store (lookupProjectionGroupStatus mainGroupId)
beforePromotion ^? _Just . #lifecyclePhase `shouldBe` Just "rebuilding-versioned"
beforePromotion ^? _Just . #servingRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v1"))
beforePromotion ^? _Just . #servingEpoch `shouldBe` Just 0
beforePromotion ^? _Just . #activeRunId
`shouldBe` Just (Just (request ^. #rebuildRunId))
beforePromotion ^? _Just . #candidateRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v2"))
promoterDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
(Tx.sql statusPromotionUpdateAndSleepSql)
)
>>= putMVar promoterDone
waitForGroupRowLock pool 50
during <- expectStore store (lookupProjectionGroupStatus mainGroupId)
during `shouldBe` beforePromotion
_ <- expectPoolUsage =<< takeMVar promoterDone
afterPromotion <- expectStore store (lookupProjectionGroupStatus mainGroupId)
afterPromotion ^? _Just . #lifecyclePhase `shouldBe` Just "serving-versioned"
afterPromotion ^? _Just . #servingRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v2"))
afterPromotion ^? _Just . #servingEpoch `shouldBe` Just 1
afterPromotion ^? _Just . #activeRunId `shouldBe` Just Nothing
afterPromotion ^? _Just . #candidateRevisionId `shouldBe` Just Nothing
it "bounds promotion behind a slow guarded reader and resumes after it exits" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom compatibleExternalReadCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "versioned-external-slow-reader" 2
let request =
versionedRequest "versioned-external-slow-reader" physicalTargets
& #cutoverLockTimeoutMs
.~ 100
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
_ <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
prepared <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
prepared ^. #promotionPrepared `shouldBe` True
readerDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
( do
Tx.sql "SET LOCAL application_name = 'keiro-slow-guarded-reader'"
rows <- Tx.statement () externalV1RowsStmt
Tx.sql "SELECT pg_sleep(1)"
pure rows
)
)
>>= putMVar readerDone
waitForNamedSession pool "keiro-slow-guarded-reader" 50
startedAt <- getCurrentTime
timedOut <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
finishedAt <- getCurrentTime
timedOut
`shouldBe` Right
(Left (VersionedCutoverDeadlineExceeded (request ^. #rebuildRunId) "promotion-group"))
diffUTCTime finishedAt startedAt `shouldSatisfy` (< 0.35)
oldRows <- expectPoolUsage =<< takeMVar readerDone
oldRows `shouldBe` [(100, 42)]
promoted <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
promoted ^. #phase `shouldBe` VersionedPromoted
it "refuses a reader whose statement snapshot crosses promotion and succeeds on retry" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom compatibleExternalReadCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request =
versionedRequest "versioned-external-late-reader" physicalTargets
& #cutoverLockTimeoutMs
.~ 2000
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
_ <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
prepared <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
prepared ^. #promotionPrepared `shouldBe` True
targetHolderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Read
( do
count <- Tx.statement () counterCountStmt
Tx.sql "SELECT pg_sleep(1)"
pure count
)
)
>>= putMVar targetHolderDone
waitForCounterReader pool 50
promoterDone <- newEmptyMVar
_ <-
forkIO
( Store.runStoreIO
store
(resumeVersionedRebuild catalog (request ^. #rebuildRunId))
>>= putMVar promoterDone
)
waitForPromotionGroupLock pool 50
lateReaderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
(Tx.statement () externalV1RowsStmt)
)
>>= putMVar lateReaderDone
threadDelay 150_000
isEmptyMVar lateReaderDone `shouldReturn` True
_ <- expectPoolUsage =<< takeMVar targetHolderDone
promotedResult <- takeMVar promoterDone
promoted <-
case promotedResult of
Left err -> expectationFailure (show err) >> error "unreachable"
Right result -> requireRight result
promoted ^. #phase `shouldBe` VersionedPromoted
crossedEpoch <- takeMVar lateReaderDone
crossedEpoch `shouldSatisfy` hasSqlState "KR001"
promotedRows <-
expectPoolUsage
=<< Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Write
(Tx.statement () externalV1RowsStmt)
)
promotedRows `shouldBe` []
runStatement store () servingCountsStmt `shouldReturn` (0, 0)
describe "schema-versioned release fault injection" $
around (withFreshDatabase fixture) $ do
it "rolls every cutover boundary back to one v1 authority and resumes the same incompatible v2 promotion" $ \connectionString ->
Store.withStore (Store.defaultConnectionSettings connectionString) $ \store ->
withPool connectionString $ \pool -> do
setupExternalBridge store
(catalog, physicalTargets) <- validatedBridgeFrom breakingExternalReadCatalog
registerBridge store catalog
runScript store servingOnlyRowsSql
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
appendVersionedEvents store "counter-release-faults" 2
let request =
versionedRequest "versioned-release-faults" physicalTargets
& #cutoverLockTimeoutMs
.~ 100
handle <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
expectStore store (reconcileExternalReadContracts catalog) >>= requireRight
captured <- driveVersionedToCutoverReady store catalog (request ^. #rebuildRunId) 10
captured ^. #phase `shouldBe` VersionedCutoverReplaying
captured ^. #capturedHead `shouldBe` GlobalPosition 2
assertPreparedV1Authority store (request ^. #rebuildRunId) False
prepared <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
prepared ^. #promotionPrepared `shouldBe` True
assertPreparedV1Authority store (request ^. #rebuildRunId) True
assertCandidateCounts store handle 2
relationHolderDone <- newEmptyMVar
_ <-
forkIO $
Pool.use
pool
( TxSessions.transactionNoRetry
TxSessions.ReadCommitted
TxSessions.Read
( do
count <- Tx.statement () counterCountStmt
Tx.sql "SELECT pg_sleep(1)"
pure count
)
)
>>= putMVar relationHolderDone
waitForCounterReader pool 50
relationRefusal <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
relationRefusal
`shouldBe` Right
(Left (VersionedCutoverDeadlineExceeded (request ^. #rebuildRunId) "target-relations"))
_ <- expectPoolUsage =<< takeMVar relationHolderDone
assertPreparedV1Authority store (request ^. #rebuildRunId) True
assertCandidateCounts store handle 2
runScript store failAfterFirstPromotionRenameSql
renameFailure <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
renameFailure `shouldSatisfy` isLeft
runScript store dropPromotionRenameFaultSql
assertPreparedV1Authority store (request ^. #rebuildRunId) True
assertCandidateCounts store handle 2
runScript store failAfterPromotionMetadataSql
metadataFailure <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
metadataFailure `shouldSatisfy` isLeft
runScript store dropPromotionMetadataFaultSql
assertPreparedV1Authority store (request ^. #rebuildRunId) True
assertCandidateCounts store handle 2
runScript store failAfterManagedWrapperSql
wrapperFailure <- Store.runStoreIO store (resumeVersionedRebuild catalog (request ^. #rebuildRunId))
wrapperFailure `shouldSatisfy` isLeft
runScript store dropManagedWrapperFaultSql
assertPreparedV1Authority store (request ^. #rebuildRunId) True
assertCandidateCounts store handle 2
promoted <- expectStore store (resumeVersionedRebuild catalog (request ^. #rebuildRunId)) >>= requireRight
promoted ^. #phase `shouldBe` VersionedPromoted
oldRead <- Store.runStoreIO store (Store.runTransaction (Tx.statement () externalV1RowsStmt))
oldRead `shouldSatisfy` hasSqlState "KR003"
runStatement store () servingCountsStmt `shouldReturn` (2, 2)
newRows <- runStatement store () externalV2RowsStmt
newRows `shouldSatisfy` (not . null)
describe "targeted stream reprojection" $
around (withFreshStore fixture) $ do
it "repairs only the selected stream and transactionally backfills redelivery evidence" $ \store -> do
catalog <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 [1, 2]
appendRepairEvents store "counter-2" 2 [7]
runStatement store () seedRepairRowsStmt
let operations = CatalogOperations.projectionCatalogOperations catalog
previewed <-
expectStore store (CatalogOperations.previewStreamReprojection operations (streamRepairRequest "counter-1"))
>>= requireRight
previewed ^. #servingRevisionId `shouldBe` identity mkProjectionRevisionId "counter-v2"
map (^. #targetId) (previewed ^. #targets) `shouldBe` [auditTargetId]
map (^. #dedupKeyId) (previewed ^. #affectedDedup)
`shouldBe` [identity mkDedupKeyId "counter-dedup"]
previewed ^. #streamVersion `shouldBe` Just (StreamVersion 2)
previewed ^. #eventCount `shouldBe` Just 2
previewed ^. #expectedDedupClaims `shouldBe` Just 2
previewed ^. #maxEvents `shouldBe` 1000
previewed ^. #eligible `shouldBe` True
previewed ^. #refusal `shouldBe` Nothing
previewed ^. #forceOperation
`shouldBe` "rebuild reproject-stream counter-group audit-owner counter-1 --page-size 1 --max-events 1000 --force"
previewed ^. #reportSchema `shouldBe` "keiro/catalog-stream-reprojection-preview/v2"
case Aeson.toJSON previewed of
Aeson.Object fields -> do
KeyMap.size fields `shouldBe` 16
all
(`KeyMap.member` fields)
[ "schema",
"groupId",
"projectionId",
"streamName",
"servingRevisionId",
"targets",
"affectedDedup",
"streamVersion",
"eventCount",
"expectedDedupClaims",
"maxEvents",
"softDeleted",
"truncateBefore",
"eligible",
"refusal",
"forceOperation"
]
`shouldBe` True
other -> expectationFailure ("expected stream-reprojection preview JSON object, got " <> show other)
repaired <-
expectStore store (CatalogOperations.reprojectCatalogStream operations (streamRepairRequest "counter-1"))
>>= requireRight
repaired ^. #reportSchema `shouldBe` "keiro/catalog-stream-reprojection-outcome/v2"
case Aeson.toJSON repaired of
Aeson.Object fields -> do
KeyMap.size fields `shouldBe` 2
all (`KeyMap.member` fields) ["schema", "repair"] `shouldBe` True
other -> expectationFailure ("expected stream-reprojection outcome JSON object, got " <> show other)
let repair = repaired ^. #repair
repair ^. #servingRevisionId `shouldBe` identity mkProjectionRevisionId "counter-v2"
repair ^. #streamVersion `shouldBe` StreamVersion 2
repair ^. #maxEvents `shouldBe` 1000
repair ^. #clearedRows `shouldBe` [StreamClearCount auditTargetId 1]
repair ^. #replayedEvents `shouldBe` 2
repair ^. #appliedEvents `shouldBe` 2
repair ^. #dedupInserted `shouldBe` 2
repair ^. #dedupExisting `shouldBe` 0
runStatement store () repairRowsStmt `shouldReturn` [(1, "3"), (2, "7")]
runStatement store () repairDedupCountStmt `shouldReturn` 2
runStatement store () repairCheckpointStmt `shouldReturn` Just 0
events <-
expectStore store (Store.readStreamForward (StreamName "counter-1") (StreamVersion 0) 10)
case Vector.toList events of
firstEvent : _ -> do
redelivery <-
expectStore
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog catalog asyncProjectionId catalogAsyncProjection firstEvent)
)
redelivery `shouldBe` CatalogAsyncDuplicate
[] -> expectationFailure "repair stream unexpectedly had no events"
runStatement store () repairRowsStmt `shouldReturn` [(1, "3"), (2, "7")]
secondRepair <-
expectStore store (reprojectStream catalog (streamRepairRequest "counter-1"))
>>= requireRight
secondRepair ^. #dedupInserted `shouldBe` 0
secondRepair ^. #dedupExisting `shouldBe` 2
runStatement store () repairRowsStmt `shouldReturn` [(1, "3"), (2, "7")]
it "refuses an oversized stream before taking the group-wide repair fence" $ \store -> do
catalog <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 (replicate 101 1)
runStatement store () seedSingleRepairRowStmt
let request = streamRepairRequest "counter-1" & #maxEvents .~ 100
operations = CatalogOperations.projectionCatalogOperations catalog
previewed <-
expectStore store (CatalogOperations.previewStreamReprojection operations request)
>>= requireRight
previewed ^. #eventCount `shouldBe` Just 101
previewed ^. #expectedDedupClaims `shouldBe` Just 101
previewed ^. #maxEvents `shouldBe` 100
previewed ^. #eligible `shouldBe` False
previewed ^. #refusal `shouldBe` Just "stream-event-limit-exceeded"
holderDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO
store
( Store.runTransaction $ do
Tx.sql "SET LOCAL application_name = 'keiro-repair-admission-group-holder'"
_ <- Tx.statement (rebuildGroupIdText mainGroupId) sanctionedRepairReaderLockStmt
Tx.sql "SELECT pg_sleep(0.75)"
)
>>= putMVar holderDone
waitForNamedSession (store ^. #pool) "keiro-repair-admission-group-holder" 50
startedAt <- getCurrentTime
refused <- expectStore store (reprojectStream catalog request)
finishedAt <- getCurrentTime
refused
`shouldBe` Left
( StreamReprojectionEventLimitExceeded
(StreamName "counter-1")
101
100
)
diffUTCTime finishedAt startedAt `shouldSatisfy` (< 0.35)
runStatement store () repairRowsStmt `shouldReturn` [(1, "999")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
holderResult <- takeMVar holderDone
holderResult `shouldBe` Right ()
it "rolls clear, replay, and dedup back when verification fails" $ \store -> do
_ <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 [1, 2]
runStatement store () seedSingleRepairRowStmt
failingCatalog <- expectValidated (streamRepairCatalogWith (repairPolicy False True))
failed <- expectStore store (reprojectStream failingCatalog (streamRepairRequest "counter-1"))
failed
`shouldSatisfy` \case
Left (StreamReprojectionVerificationFailed "forced verification failure") -> True
_ -> False
runStatement store () repairRowsStmt `shouldReturn` [(1, "999")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
it "rolls back clearer and decode failures without dedup evidence" $ \store -> do
catalog <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 [1]
runStatement store () seedSingleRepairRowStmt
clearerCatalog <- expectValidated (streamRepairCatalogWith clearFailureRepairPolicy)
clearerFailed <- expectStore store (reprojectStream clearerCatalog (streamRepairRequest "counter-1"))
clearerFailed `shouldBe` Left (StreamReprojectionClearFailed "forced clearer failure")
runStatement store () repairRowsStmt `shouldReturn` [(1, "999")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
appendMalformedRepairEvent store "counter-1"
decodeFailed <- expectStore store (reprojectStream catalog (streamRepairRequest "counter-1"))
decodeFailed
`shouldSatisfy` \case
Left (StreamReprojectionDecodeFailed (StreamVersion 2) _) -> True
_ -> False
runStatement store () repairRowsStmt `shouldReturn` [(1, "999")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
it "refuses truncated and soft-deleted streams before target mutation" $ \store -> do
catalog <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 [1]
appendRepairEvents store "counter-2" 2 [7]
runStatement store () seedRepairRowsStmt
_ <- expectStore store (Store.setStreamTruncateBefore (StreamName "counter-1") (StreamVersion 1))
_ <- expectStore store (Store.softDeleteStream (StreamName "counter-2"))
truncated <- expectStore store (reprojectStream catalog (streamRepairRequest "counter-1"))
truncated
`shouldSatisfy` \case
Left (StreamReprojectionTruncated (StreamName "counter-1") (StreamVersion 1)) -> True
_ -> False
deleted <- expectStore store (reprojectStream catalog (streamRepairRequest "counter-2"))
deleted
`shouldSatisfy` \case
Left (StreamReprojectionSoftDeleted (StreamName "counter-2")) -> True
_ -> False
runStatement store () repairRowsStmt `shouldReturn` [(1, "999"), (2, "7")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
it "refuses an active rebuild and an unavailable persisted serving revision" $ \store -> do
setupBridge store
(catalog, physicalTargets) <- validatedBridgeFrom streamRepairCatalog
registerBridge store catalog
appendRepairEvents store "counter-1" 1 [1]
let activeRequest = versionedRequest "stream-repair-active-rebuild" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog activeRequest) >>= requireRight
active <- expectStore store (reprojectStream catalog (streamRepairRequest "counter-1"))
active
`shouldBe` Left (StreamReprojectionActiveRebuild mainGroupId (activeRequest ^. #rebuildRunId))
abandoned <- expectStore store (abandonVersionedRebuild (activeRequest ^. #rebuildRunId))
abandoned `shouldSatisfy` isRight
unavailableCatalog <-
expectValidated
( streamRepairCatalog
& #projectionRevisions
%~ filter
(\revision -> revision ^. #revisionId /= identity mkProjectionRevisionId "counter-v1")
)
unavailableSlice <-
maybe
(expectationFailure "unavailable-revision catalog has no group slice" >> error "unreachable")
pure
(Catalog.groupSliceFingerprint unavailableCatalog mainGroupId)
runStatement store (groupSliceFingerprintText unavailableSlice) forceGroupSliceStmt
missing <- expectStore store (reprojectStream unavailableCatalog (streamRepairRequest "counter-1"))
missing
`shouldBe` Left
( StreamReprojectionServingRevisionUnavailable
mainGroupId
(identity mkProjectionRevisionId "counter-v1")
)
it "waits for an append-first catalog writer and replays its committed event" $ \store -> do
catalog <- prepareStreamRepair store streamRepairCatalog
appendRepairEvents store "counter-1" 1 [1]
appendRepairEvents store "counter-2" 2 [7]
runStatement store () seedRepairRowsStmt
prepared <-
StoreTransaction.prepareEventsIO
[ EventData
{ eventId = Nothing,
eventType = EventType "RepairDelta",
payload = Aeson.toJSON ([1, 5] :: [Int64]),
metadata = Nothing,
causationId = Nothing,
correlationId = Nothing
}
]
now <- getCurrentTime
writerDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO
store
( Store.runTransaction $ do
appended <-
StoreTransaction.appendToStreamTx
(StreamName "counter-1")
(ExactVersion (StreamVersion 1))
prepared
now
case appended of
Left conflict -> Tx.condemn >> pure (Left conflict)
Right result -> do
Tx.sql "SELECT pg_sleep(0.75)"
_ <- Tx.statement (rebuildGroupIdText mainGroupId) sanctionedRepairReaderLockStmt
pure (Right result)
)
>>= putMVar writerDone
threadDelay 150_000
repairDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO store (reprojectStream catalog (streamRepairRequest "counter-1"))
>>= putMVar repairDone
threadDelay 150_000
isEmptyMVar repairDone `shouldReturn` True
writerResult <- takeMVar writerDone
writerResult `shouldSatisfy` \case Right (Right _) -> True; _ -> False
repairResult <- takeMVar repairDone
case repairResult of
Right (Right report) -> report ^. #streamVersion `shouldBe` StreamVersion 2
other -> expectationFailure ("append-first repair failed: " <> show other)
runStatement store () repairRowsStmt `shouldReturn` [(1, "6"), (2, "7")]
runStatement store () repairDedupCountStmt `shouldReturn` 2
it "holds the group writer fence until one-stream repair commits" $ \store -> do
_ <- prepareStreamRepair store streamRepairCatalog
slowCatalog <- expectValidated (streamRepairCatalogWith (repairPolicy True False))
appendRepairEvents store "counter-1" 1 [1]
appendRepairEvents store "counter-2" 2 [7]
runStatement store () seedRepairRowsStmt
secondStream <-
expectStore store (Store.readStreamForward (StreamName "counter-2") (StreamVersion 0) 10)
writerEvent <-
case Vector.toList secondStream of
event : _ -> pure event
[] -> expectationFailure "writer stream unexpectedly empty" >> error "unreachable"
repairDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO store (reprojectStream slowCatalog (streamRepairRequest "counter-1"))
>>= putMVar repairDone
threadDelay 150_000
readerDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO
store
( Store.runTransaction $ do
_ <- Tx.statement (rebuildGroupIdText mainGroupId) sanctionedRepairReaderLockStmt
Tx.statement (1 :: Int64) repairDetailByIdStmt
)
>>= putMVar readerDone
writerDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO
store
( Store.runTransaction
(applyAsyncProjectionFromCatalog slowCatalog asyncProjectionId catalogAsyncProjection writerEvent)
)
>>= putMVar writerDone
threadDelay 150_000
isEmptyMVar writerDone `shouldReturn` True
isEmptyMVar readerDone `shouldReturn` True
runStatement store () repairRowsStmt `shouldReturn` [(1, "999"), (2, "7")]
repairResult <- takeMVar repairDone
repairResult `shouldSatisfy` \case Right (Right _) -> True; _ -> False
readerResult <- takeMVar readerDone
readerResult `shouldBe` Right "1"
writerResult <- takeMVar writerDone
writerResult `shouldBe` Right CatalogAsyncApplied
it "serializes hard deletion behind the stream guard" $ \store -> do
_ <- prepareStreamRepair store streamRepairCatalog
slowCatalog <- expectValidated (streamRepairCatalogWith (repairPolicy True False))
appendRepairEvents store "counter-1" 1 [1]
runStatement store () seedSingleRepairRowStmt
repairDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO store (reprojectStream slowCatalog (streamRepairRequest "counter-1"))
>>= putMVar repairDone
_ <- waitForApplicationPid (store ^. #pool) "keiro-stream-repair-after-clear" 50
deleteDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO store (Store.hardDeleteStream (StreamName "counter-1"))
>>= putMVar deleteDone
threadDelay 150_000
isEmptyMVar deleteDone `shouldReturn` True
repairResult <- takeMVar repairDone
repairResult `shouldSatisfy` \case Right (Right _) -> True; _ -> False
deleted <- takeMVar deleteDone
deleted `shouldSatisfy` \case Right (Just _) -> True; _ -> False
runStatement store () repairRowsStmt `shouldReturn` [(1, "1")]
missing <- expectStore store (reprojectStream slowCatalog (streamRepairRequest "counter-1"))
missing
`shouldSatisfy` \case
Left (StreamReprojectionHistoryUnavailable _) -> True
_ -> False
it "rolls target and dedup changes back when the repair backend is interrupted" $ \store -> do
_ <- prepareStreamRepair store streamRepairCatalog
slowCatalog <- expectValidated (streamRepairCatalogWith (repairPolicy True False))
appendRepairEvents store "counter-1" 1 [1]
runStatement store () seedSingleRepairRowStmt
repairDone <- newEmptyMVar
_ <-
forkIO $
Store.runStoreIO store (reprojectStream slowCatalog (streamRepairRequest "counter-1"))
>>= putMVar repairDone
backendPid <- waitForApplicationPid (store ^. #pool) "keiro-stream-repair-after-clear" 50
terminated <-
expectPoolUsage
=<< Pool.use (store ^. #pool) (Session.statement backendPid terminateBackendStmt)
terminated `shouldBe` True
interrupted <- takeMVar repairDone
interrupted `shouldSatisfy` isLeft
runStatement store () repairRowsStmt `shouldReturn` [(1, "999")]
runStatement store () repairDedupCountStmt `shouldReturn` 0
validatedBridge :: IO (ValidatedProjectionCatalog, PhysicalTargets)
validatedBridge = validatedBridgeFrom runtimeBridgeCatalog
validatedBridgeFrom :: ProjectionCatalog -> IO (ValidatedProjectionCatalog, PhysicalTargets)
validatedBridgeFrom catalogDefinition = do
catalog <-
case validateProjectionCatalog catalogDefinition of
Failure diagnostics -> expectationFailure (show diagnostics) >> error "unreachable"
Success validated -> pure validated
physicalTargets <-
case mkPhysicalTargets
[counterTargetId, auditTargetId]
( Map.fromList
[ (counterTargetId, QualifiedTable "app" "counter"),
(auditTargetId, QualifiedTable "app" "counter_audit")
]
) of
Left errors -> expectationFailure (show errors) >> error "unreachable"
Right targets -> pure targets
pure (catalog, physicalTargets)
runtimeBridgeCatalog :: ProjectionCatalog
runtimeBridgeCatalog =
bridgeCatalog
{ projectionRevisions =
[ runtimeRevision "v1" bridgeRevisionV1,
runtimeRevision "v2" bridgeRevisionV2
],
externalReadContracts = []
}
streamRepairCatalog :: ProjectionCatalog
streamRepairCatalog = streamRepairCatalogWith (repairPolicy False False)
streamRepairCatalogWith :: StreamScopedReplay -> ProjectionCatalog
streamRepairCatalogWith policy =
runtimeBridgeCatalog
{ projectionRevisions =
[ revision & #streamScopedReplays .~ [policy]
| revision <- runtimeBridgeCatalog ^. #projectionRevisions
]
}
repairPolicy :: Bool -> Bool -> StreamScopedReplay
repairPolicy pauseAfterClear failVerification =
StreamScopedReplay
{ streamProjectionId = asyncProjectionId,
streamOwnedTargets = auditTargetId :| [],
clearerId = "runtime-audit/clear-stream",
clearerVersion = 1,
clearStreamRows = \physicalTargets streamName ->
case repairAggregateId streamName of
Left detail -> pure (Left detail)
Right aggregateId -> do
let auditTable = requireTargetFrom physicalTargets auditTargetId
cleared <- Tx.statement aggregateId (deleteRepairRowStmt auditTable)
when pauseAfterClear $ do
Tx.sql "SET LOCAL application_name = 'keiro-stream-repair-after-clear'"
Tx.sql "SELECT pg_sleep(0.75)"
pure (Right [StreamClearCount auditTargetId cleared]),
streamReplayId = "runtime-audit/replay-stream",
streamReplayVersion = 1,
replayStreamEvent = \physicalTargets repairRecorded ->
case Aeson.fromJSON (repairRecorded ^. #payload) :: Aeson.Result [Int64] of
Aeson.Success [aggregateId, delta] -> do
let auditTable = requireTargetFrom physicalTargets auditTargetId
StreamVersion sourceVersion = repairRecorded ^. #streamVersion
Tx.statement (aggregateId, delta, sourceVersion) (upsertRepairRowStmt auditTable)
pure (Right True)
Aeson.Success _ -> pure (Left (ReplayDecodeError "repair payload must contain aggregate id and delta"))
Aeson.Error detail -> pure (Left (ReplayDecodeError (Text.pack detail))),
streamVerificationId = "runtime-audit/verify-stream",
streamVerificationVersion = 1,
verifyStreamRows = \physicalTargets streamName ->
if failVerification
then pure (Left "forced verification failure")
else case repairAggregateId streamName of
Left detail -> pure (Left detail)
Right aggregateId -> do
let auditTable = requireTargetFrom physicalTargets auditTargetId
rows <- Tx.statement aggregateId (countRepairRowsStmt auditTable)
pure
( if rows == 1
then Right ()
else Left "repair must leave exactly one aggregate row"
),
affectedAsyncDedup = [identity mkDedupKeyId "counter-dedup"],
claimSite = identity mkClaimSite "versioned:runtime-audit-stream-repair"
}
clearFailureRepairPolicy :: StreamScopedReplay
clearFailureRepairPolicy =
(repairPolicy False False)
{ clearStreamRows = \_ _ -> pure (Left "forced clearer failure")
}
prepareStreamRepair :: Store.KirokuStore -> ProjectionCatalog -> IO ValidatedProjectionCatalog
prepareStreamRepair store catalogDefinition = do
setupBridge store
(catalog, physicalTargets) <- validatedBridgeFrom catalogDefinition
registerBridge store catalog
runStatement store ("catalog-async-subscription", 0) upsertSubscriptionCursorStmt
let request = versionedRequest "stream-repair-bootstrap" physicalTargets
_ <- expectStore store (beginVersionedRebuild catalog request) >>= requireRight
promoted <- driveVersionedToPromotion store catalog (request ^. #rebuildRunId) 10
promoted ^. #phase `shouldBe` VersionedPromoted
pure catalog
expectValidated :: ProjectionCatalog -> IO ValidatedProjectionCatalog
expectValidated catalogDefinition =
case validateProjectionCatalog catalogDefinition of
Failure diagnostics -> expectationFailure (show diagnostics) >> error "unreachable"
Success catalog -> pure catalog
streamRepairRequest :: Text -> StreamReprojectionRequest
streamRepairRequest stream =
StreamReprojectionRequest
{ rebuildGroupId = mainGroupId,
projectionId = asyncProjectionId,
streamName = StreamName stream,
pageSize = 1,
maxEvents = 1000
}
repairAggregateId :: StreamName -> Either Text Int64
repairAggregateId = \case
StreamName "counter-1" -> Right 1
StreamName "counter-2" -> Right 2
StreamName name -> Left ("unexpected repair stream: " <> name)
requireTargetFrom :: PhysicalTargets -> TargetId -> QualifiedTable
requireTargetFrom physicalTargets targetId =
fromMaybe
(error ("missing repair target: " <> Text.unpack (targetIdText targetId)))
(resolvePhysicalTarget targetId physicalTargets)
compatibleExternalReadCatalog :: ProjectionCatalog
compatibleExternalReadCatalog =
runtimeBridgeCatalog
{ externalReadContracts = [counterReadContract]
}
breakingExternalReadCatalog :: ProjectionCatalog
breakingExternalReadCatalog = externalReadCatalogWith counterV1BreakingContract
compatibilityImplementationCatalog :: ProjectionCatalog
compatibilityImplementationCatalog = externalReadCatalogWith counterV1CompatibilityContract
externalReadCatalogWith :: ExternalReadContract -> ProjectionCatalog
externalReadCatalogWith v1Contract =
runtimeBridgeCatalog
{ queryModels = runtimeBridgeCatalog ^. #queryModels <> [SomeQueryModelBinding counterV2Binding],
externalReadContracts = [v1Contract, counterV2Contract]
}
counterV2Binding :: QueryModelBinding Text ()
counterV2Binding =
counterBinding
{ queryModelId = identity mkQueryModelId "catalog-counter-query-v2",
readModel =
(counterBinding ^. #readModel)
{ name = "catalog-counter-query-v2",
version = 2,
shapeHash = "catalog-counter-query-v2"
}
}
counterV1BreakingContract :: ExternalReadContract
counterV1BreakingContract =
counterReadContract
& #compatibleRevisions
.~ (identity mkProjectionRevisionId "counter-v1" :| [])
counterV1CompatibilityContract :: ExternalReadContract
counterV1CompatibilityContract =
KeyedExternalRead
{ readContractId = counterReadContract ^. #readContractId,
contractVersion = counterReadContract ^. #contractVersion,
queryModelId = counterReadContract ^. #queryModelId,
arguments = [],
resultContractType = counterReadContract ^. #resultContractType,
privateImplementation = QualifiedFunction "app_private" "counter_v1_compat",
privateImplementationVersion = 1,
resultShapeHash = counterReadContract ^. #resultShapeHash,
compatibleRevisions =
identity mkProjectionRevisionId "counter-v1"
:| [identity mkProjectionRevisionId "counter-v2"],
surfaceGeneration = 1,
claimSite = identity mkClaimSite "versioned:counter-v1-compatibility"
}
counterV2Contract :: ExternalReadContract
counterV2Contract =
AllRowsExternalRead
{ readContractId = counterReadContract ^. #readContractId,
contractVersion = ExternalReadContractVersion 2,
queryModelId = identity mkQueryModelId "catalog-counter-query-v2",
resultContractType = QualifiedSqlType "app_contract" "counter_row_v2",
resultShapeHash = "catalog-counter-query-v2",
compatibleRevisions = identity mkProjectionRevisionId "counter-v2" :| [],
surfaceGeneration = 2,
claimSite = identity mkClaimSite "versioned:counter-reader-v2"
}
runtimeV1OnlyCatalog :: ProjectionCatalog
runtimeV1OnlyCatalog =
runtimeBridgeCatalog
{ projectionRevisions = [runtimeRevision "v1" bridgeRevisionV1],
externalReadContracts = []
}
cloneBridgeCatalog :: ProjectionCatalog
cloneBridgeCatalog =
runtimeBridgeCatalog
{ projectionRevisions =
[ runtimeRevision "v1" bridgeRevisionV1,
runtimeRevision "v1" bridgeRevisionV2
]
}
identityCloneBridgeCatalog :: ProjectionCatalog
identityCloneBridgeCatalog =
cloneBridgeCatalog
{ projectionRevisions =
[ revision
& #targetProvisioners
%~ Map.adjust identityCloneProvisioner counterTargetId
| revision <- cloneBridgeCatalog ^. #projectionRevisions
]
}
identityCloneProvisioner :: TargetProvisioner -> TargetProvisioner
identityCloneProvisioner provisioner =
provisioner
{ validateTarget = Just (validateRuntimeTarget counterTargetId "v1" identityCloneObjects),
promotionObjectNames = identityCloneObjects
}
identityCloneObjects :: [PromotionObjectName]
identityCloneObjects =
[ PromotionObjectName PromotionConstraint "counter_pkey__clone" "counter_pkey",
PromotionObjectName PromotionOwnedSequence "counter_id_seq__clone" "counter_id_seq"
]
raceBridgeCatalog :: ProjectionCatalog
raceBridgeCatalog =
replaceCandidateProvisionerInCatalog
counterTargetId
(\provisioner -> provisioner {validateTarget = Just validateRaceCounter})
runtimeBridgeCatalog
validateRaceCounter ::
TargetProvisioningContext ->
Tx.Transaction (Either [TargetSchemaViolation] TargetSchemaEvidence)
validateRaceCounter targetContext = do
baseline <- validateRuntimeTarget counterTargetId "v2" promotionObjects targetContext
rogue <- Tx.statement (targetContext ^. #stagingTable) rogueColumnStmt
pure $
baseline <&> \evidence ->
if rogue
then evidence & #observedShapeFingerprint %~ (<> ":rogue")
else evidence
where
promotionObjects =
[PromotionObjectName PromotionIndex "counter_total_idx__v2" "counter_total_idx"]
runtimeRevision :: Text -> ProjectionRevision -> ProjectionRevision
runtimeRevision schema revision =
revision
& #targetProvisioners
.~ Map.fromList
[ (counterTargetId, counterProvisioner schema),
(auditTargetId, auditProvisioner schema)
]
& #liveHandlers
.~ [ RevisionLiveHandler
("runtime-inline-live-" <> schema)
1
(RevisionInlineDelivery inlineProjectionId "catalog-inline")
[counterTargetId]
(applyRuntimeCounter schema),
RevisionLiveHandler
("runtime-async-live-" <> schema)
1
( RevisionSubscriptionDelivery
asyncProjectionId
(identity mkSubscriptionId "counter-subscription")
(identity mkDedupKeyId "counter-dedup")
)
[auditTargetId]
(applyRuntimeAudit schema)
]
& #replayAdapters
.~ [ RevisionReplayAdapter
("runtime-replay-" <> schema)
1
[counterTargetId, auditTargetId]
( \physicalTargets event -> do
applyRuntimeLive schema physicalTargets event
pure (Right True)
)
]
applyRuntimeLive :: Text -> PhysicalTargets -> RecordedEvent -> Tx.Transaction ()
applyRuntimeLive schema physicalTargets event = do
applyRuntimeCounter schema physicalTargets event
applyRuntimeAudit schema physicalTargets event
applyRuntimeCounter :: Text -> PhysicalTargets -> RecordedEvent -> Tx.Transaction ()
applyRuntimeCounter schema physicalTargets event = do
let counterTable = requireTarget counterTargetId
eventPosition = positionValue (event ^. #globalPosition)
counterQualified = qualifyTable (counterTable ^. #schemaName) (counterTable ^. #tableName)
if schema == "v1"
then
Tx.sql
( Text.Encoding.encodeUtf8
( "INSERT INTO "
<> counterQualified
<> " (id, total) VALUES ("
<> Text.pack (show eventPosition)
<> ", 10)"
)
)
else
Tx.sql
( Text.Encoding.encodeUtf8
( "INSERT INTO "
<> counterQualified
<> " (id, subtotal, tax) VALUES ("
<> Text.pack (show eventPosition)
<> ", 8, 2)"
)
)
where
requireTarget targetId =
fromMaybe
(error ("runtime live handler missing target " <> show targetId))
(resolvePhysicalTarget targetId physicalTargets)
applyRuntimeAudit :: Text -> PhysicalTargets -> RecordedEvent -> Tx.Transaction ()
applyRuntimeAudit schema physicalTargets event = do
let auditTable =
fromMaybe
(error "runtime async live handler missing audit target")
(resolvePhysicalTarget auditTargetId physicalTargets)
eventPosition = positionValue (event ^. #globalPosition)
auditQualified = qualifyTable (auditTable ^. #schemaName) (auditTable ^. #tableName)
if schema == "v1"
then
Tx.sql
( Text.Encoding.encodeUtf8
( "INSERT INTO "
<> auditQualified
<> " (id, detail) VALUES ("
<> Text.pack (show eventPosition)
<> ", 'v1')"
)
)
else
Tx.sql
( Text.Encoding.encodeUtf8
( "INSERT INTO "
<> auditQualified
<> " (id, detail, source_position) VALUES ("
<> Text.pack (show eventPosition)
<> ", 'v2', "
<> Text.pack (show eventPosition)
<> ")"
)
)
positionValue :: GlobalPosition -> Int64
positionValue (GlobalPosition value) = value
recorded :: Int64 -> RecordedEvent
recorded value =
RecordedEvent
{ eventId = EventId (UUID.fromWords64 7 (fromIntegral value)),
eventType = EventType "VersionedDispatch",
streamVersion = StreamVersion value,
globalPosition = GlobalPosition value,
originalStreamId = StreamId value,
originalVersion = StreamVersion value,
payload = Aeson.Null,
metadata = Just (Aeson.object []),
causationId = Nothing,
correlationId = Nothing,
createdAt = UTCTime (ModifiedJulianDay 0) (secondsToDiffTime 0)
}
counterProvisioner :: Text -> TargetProvisioner
counterProvisioner schema =
TargetProvisioner
{ provisionerId = "runtime-counter-" <> schema,
provisionerVersion = 1,
schemaVersion = TargetSchemaVersion schema,
expectedShapeId = "runtime-counter-shape-" <> schema,
provisionTarget =
if schema == "v1"
then \_ -> pure ()
else createV2Counter,
validatorId = "runtime-counter-validator-" <> schema,
validatorVersion = 1,
validateTarget = Just (validateRuntimeTarget counterTargetId schema promotionObjects),
promotionObjectNames = promotionObjects
}
where
promotionObjects =
[PromotionObjectName PromotionIndex "counter_total_idx__v2" "counter_total_idx" | schema == "v2"]
auditProvisioner :: Text -> TargetProvisioner
auditProvisioner schema =
TargetProvisioner
{ provisionerId = "runtime-audit-" <> schema,
provisionerVersion = 1,
schemaVersion = TargetSchemaVersion schema,
expectedShapeId = "runtime-audit-shape-" <> schema,
provisionTarget =
if schema == "v1"
then \_ -> pure ()
else createV2Audit,
validatorId = "runtime-audit-validator-" <> schema,
validatorVersion = 1,
validateTarget = Just (validateRuntimeTarget auditTargetId schema []),
promotionObjectNames = []
}
validateRuntimeTarget ::
TargetId ->
Text ->
[PromotionObjectName] ->
TargetProvisioningContext ->
Tx.Transaction (Either [TargetSchemaViolation] TargetSchemaEvidence)
validateRuntimeTarget targetId schema promotionObjects targetContext = do
maybeOid <- Tx.statement (targetContext ^. #stagingTable) relationOidStmt
pure $ case maybeOid of
Nothing -> Left [TargetSchemaViolation "relation.missing" (targetIdText targetId)]
Just oid ->
Right
TargetSchemaEvidence
{ relationOid = oid,
observedShapeFingerprint = "runtime-" <> targetIdText targetId <> "-shape-" <> schema,
observedPromotionObjects = promotionObjects,
catalogSnapshot = "runtime-catalog-snapshot-" <> schema
}
createV2Counter :: TargetProvisioningContext -> Tx.Transaction ()
createV2Counter targetContext = do
let table = targetContext ^. #stagingTable
qualified = qualifyTable (table ^. #schemaName) (table ^. #tableName)
Tx.sql
( Text.Encoding.encodeUtf8
( "CREATE TABLE "
<> qualified
<> " (id bigint PRIMARY KEY, subtotal bigint NOT NULL, tax bigint NOT NULL, total bigint GENERATED ALWAYS AS (subtotal + tax) STORED)"
)
)
Tx.sql
( Text.Encoding.encodeUtf8
( "CREATE INDEX \"counter_total_idx__v2\" ON "
<> qualified
<> " (total)"
)
)
createV2Audit :: TargetProvisioningContext -> Tx.Transaction ()
createV2Audit targetContext = do
let table = targetContext ^. #stagingTable
Tx.sql
( Text.Encoding.encodeUtf8
( "CREATE TABLE "
<> qualifyTable (table ^. #schemaName) (table ^. #tableName)
<> " (id bigint PRIMARY KEY, detail text NOT NULL, source_position bigint NOT NULL)"
)
)
replaceCandidateProvisionerInCatalog ::
TargetId ->
(TargetProvisioner -> TargetProvisioner) ->
ProjectionCatalog ->
ProjectionCatalog
replaceCandidateProvisionerInCatalog targetId update catalog =
catalog
{ projectionRevisions =
[ if revision ^. #revisionId == identity mkProjectionRevisionId "counter-v2"
then revision & #targetProvisioners %~ Map.adjust update targetId
else revision
| revision <- catalog ^. #projectionRevisions
]
}
registerBridge :: Store.KirokuStore -> ValidatedProjectionCatalog -> IO ()
registerBridge store catalog = do
result <- expectStore store (registerProjectionCatalog catalog)
case result of
Left err -> expectationFailure (show err)
Right _ -> pure ()
versionedRequest :: Text -> PhysicalTargets -> VersionedRebuildRequest
versionedRequest identityText = rebuildRequestFor (run identityText)
rebuildRequestFor :: RebuildRunId -> PhysicalTargets -> VersionedRebuildRequest
rebuildRequestFor runId physicalTargets =
VersionedRebuildRequest
{ rebuildRunId = runId,
rebuildGroupId = mainGroupId,
servingRevisionId = identity mkProjectionRevisionId "counter-v1",
candidateRevisionId = identity mkProjectionRevisionId "counter-v2",
servingTargets = physicalTargets,
targetMode = ApplicationProvisioned,
replayPageSize = 2,
cutoverThreshold = 10,
cutoverLockTimeoutMs = 2_000,
promotionDedupLimit = 1_000_000,
retentionLeaseRequest = retentionRequest runId,
requestedBy = "versioned-rebuild-spec",
requestReason = "exercise M3 lifecycle persistence"
}
retentionRequest :: RebuildRunId -> HistoryRetentionLeaseRequest
retentionRequest runId =
HistoryRetentionLeaseRequest
{ owner = requireIdentity (mkHistoryRetentionLeaseOwner ("keiro-rebuild/" <> rebuildRunIdText runId)),
reason = requireIdentity (mkHistoryRetentionLeaseReason "schema-versioned projection rebuild"),
duration = requireIdentity (mkHistoryRetentionLeaseDuration (secondsToDiffTime 600))
}
candidateTable :: RebuildRunId -> TargetId -> QualifiedTable
candidateTable runId targetId =
QualifiedTable "app" ("keiro_g_" <> Text.filter (/= '-') (UUID.toText generationId))
where
generationId =
UUID.V5.generateNamed
UUID.V5.namespaceURL
( ByteString.unpack
( Text.Encoding.encodeUtf8
( Text.intercalate
"\NUL"
[ "keiro/versioned-candidate-generation/v1",
rebuildRunIdText runId,
rebuildGroupIdText mainGroupId,
targetIdText targetId
]
)
)
)
setupBridge :: Store.KirokuStore -> IO ()
setupBridge store = runScript store bridgeSql
setupExternalBridge :: Store.KirokuStore -> IO ()
setupExternalBridge store = runScript store externalBridgeSql
bridgeSql :: ByteString
bridgeSql =
"""
CREATE SCHEMA app;
CREATE TABLE app.counter (
id bigint PRIMARY KEY,
total bigint NOT NULL
);
CREATE TABLE app.counter_audit (
id bigint PRIMARY KEY,
detail text NOT NULL
);
"""
externalBridgeSql :: ByteString
externalBridgeSql =
bridgeSql
<> """
CREATE SCHEMA app_contract;
CREATE TYPE app_contract.counter_row_v1 AS (id bigint, total bigint);
CREATE TYPE app_contract.counter_row_v2 AS
(id bigint, subtotal bigint, tax bigint, total bigint);
CREATE SCHEMA app_private;
CREATE FUNCTION app_private.counter_v1_compat()
RETURNS SETOF app_contract.counter_row_v1
LANGUAGE plpgsql
STABLE
AS $compatibility$
BEGIN
RETURN QUERY EXECUTE
'SELECT counter.id, counter.total FROM app.counter AS counter ORDER BY counter.id';
END
$compatibility$;
"""
identityBridgeSql :: ByteString
identityBridgeSql =
"""
CREATE SCHEMA app;
CREATE TABLE app.counter (
id bigint GENERATED BY DEFAULT AS IDENTITY,
total bigint NOT NULL,
CONSTRAINT counter_pkey PRIMARY KEY (id)
);
CREATE TABLE app.counter_audit (
id bigint PRIMARY KEY,
detail text NOT NULL
);
"""
cloneTriggerSql :: ByteString
cloneTriggerSql =
"""
CREATE FUNCTION app.clone_refusal_trigger() RETURNS trigger
LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END $$;
CREATE TRIGGER clone_refusal
BEFORE INSERT ON app.counter
FOR EACH ROW EXECUTE FUNCTION app.clone_refusal_trigger();
"""
retiredReaderSql :: ByteString
retiredReaderSql =
"CREATE VIEW app.retired_counter_reader AS SELECT id, total FROM app.counter"
servingOnlyRowsSql :: ByteString
servingOnlyRowsSql =
"""
INSERT INTO app.counter (id, total) VALUES (100, 42);
INSERT INTO app.counter_audit (id, detail) VALUES (100, 'serving-v1');
"""
failAfterFirstPromotionRenameSql :: ByteString
failAfterFirstPromotionRenameSql =
"""
CREATE FUNCTION public.keiro_test_fail_after_promotion_rename()
RETURNS event_trigger
LANGUAGE plpgsql
AS $$
BEGIN
RAISE EXCEPTION 'injected failure after a promotion rename';
END
$$;
CREATE EVENT TRIGGER keiro_test_fail_after_promotion_rename
ON ddl_command_end
WHEN TAG IN ('ALTER TABLE')
EXECUTE FUNCTION public.keiro_test_fail_after_promotion_rename();
"""
dropPromotionRenameFaultSql :: ByteString
dropPromotionRenameFaultSql =
"""
DROP EVENT TRIGGER keiro_test_fail_after_promotion_rename;
DROP FUNCTION public.keiro_test_fail_after_promotion_rename();
"""
failAfterPromotionMetadataSql :: ByteString
failAfterPromotionMetadataSql =
"""
CREATE FUNCTION public.keiro_test_fail_after_promotion_metadata()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
RAISE EXCEPTION 'injected failure after the promotion metadata transition';
END
$$;
CREATE TRIGGER keiro_test_fail_after_promotion_metadata
AFTER UPDATE ON keiro.keiro_projection_rebuild_groups
FOR EACH ROW
WHEN (OLD.status = 'cutover-versioned' AND NEW.status = 'serving-versioned')
EXECUTE FUNCTION public.keiro_test_fail_after_promotion_metadata();
"""
dropPromotionMetadataFaultSql :: ByteString
dropPromotionMetadataFaultSql =
"""
DROP TRIGGER keiro_test_fail_after_promotion_metadata
ON keiro.keiro_projection_rebuild_groups;
DROP FUNCTION public.keiro_test_fail_after_promotion_metadata();
"""
failAfterManagedWrapperSql :: ByteString
failAfterManagedWrapperSql =
"""
CREATE FUNCTION public.keiro_test_fail_after_managed_wrapper()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
RAISE EXCEPTION 'injected failure after managed wrapper reconciliation';
END
$$;
CREATE TRIGGER keiro_test_fail_after_managed_wrapper
AFTER INSERT OR UPDATE ON keiro.keiro_managed_read_objects
FOR EACH ROW
WHEN (NEW.object_kind = 'wrapper-function')
EXECUTE FUNCTION public.keiro_test_fail_after_managed_wrapper();
"""
dropManagedWrapperFaultSql :: ByteString
dropManagedWrapperFaultSql =
"""
DROP TRIGGER keiro_test_fail_after_managed_wrapper
ON keiro.keiro_managed_read_objects;
DROP FUNCTION public.keiro_test_fail_after_managed_wrapper();
"""
promoteDispatchMetadataSql :: ByteString
promoteDispatchMetadataSql =
"""
UPDATE keiro.keiro_projection_target_generations
SET lifecycle = 'retired', retired_at = now()
WHERE group_id = 'counter-group' AND lifecycle = 'serving';
UPDATE keiro.keiro_projection_target_generations AS generations
SET lifecycle = 'serving', served_at = now()
FROM keiro.keiro_projection_rebuild_run_targets AS targets
WHERE targets.run_id = 'versioned-dispatch'
AND targets.candidate_generation_id = generations.generation_id
AND generations.lifecycle = 'staging';
UPDATE keiro.keiro_projection_rebuild_runs
SET status = 'promoted', history_retention_released_at = now(), updated_at = now()
WHERE run_id = 'versioned-dispatch';
UPDATE keiro.keiro_projection_rebuild_groups
SET status = 'serving-versioned',
active_run_id = NULL,
serving_revision_id = 'counter-v2',
serving_epoch = serving_epoch + 1,
reads_allowed = TRUE,
writes_allowed = TRUE,
completed_at = now(),
updated_at = now()
WHERE group_id = 'counter-group';
"""
runScript :: Store.KirokuStore -> ByteString -> IO ()
runScript store sql = expectStore store (Store.runTransaction (Tx.sql sql))
runStatement :: Store.KirokuStore -> params -> Statement params result -> IO result
runStatement store params statement =
expectStore store (Store.runTransaction (Tx.statement params statement))
withPool :: Text -> (Pool.Pool -> IO a) -> IO a
withPool connectionString =
bracket
( Pool.acquire $
PoolConfig.settings
[ PoolConfig.staticConnectionSettings (ConnectionSettings.connectionString connectionString),
PoolConfig.size 4
]
)
Pool.release
expectPoolUsage :: (Show error) => Either error value -> IO value
expectPoolUsage = \case
Left err -> expectationFailure ("database action failed: " <> show err) >> error "unreachable"
Right value -> pure value
waitForCounterReader :: Pool.Pool -> Int -> IO ()
waitForCounterReader _ 0 = expectationFailure "reader did not acquire ACCESS SHARE in time"
waitForCounterReader pool remaining = do
locked <-
expectPoolUsage
=<< Pool.use pool (TxSessions.transactionNoRetry TxSessions.ReadCommitted TxSessions.Read (Tx.statement () counterAccessShareStmt))
if locked
then pure ()
else threadDelay 20_000 >> waitForCounterReader pool (remaining - 1)
waitForAuditReader :: Pool.Pool -> Int -> IO ()
waitForAuditReader _ 0 = expectationFailure "audit reader did not acquire ACCESS SHARE in time"
waitForAuditReader pool remaining = do
locked <-
expectPoolUsage
=<< Pool.use pool (TxSessions.transactionNoRetry TxSessions.ReadCommitted TxSessions.Read (Tx.statement () auditAccessShareStmt))
if locked
then pure ()
else threadDelay 20_000 >> waitForAuditReader pool (remaining - 1)
waitForGroupRowLock :: Pool.Pool -> Int -> IO ()
waitForGroupRowLock _ 0 = expectationFailure "group row holder did not acquire its lock in time"
waitForGroupRowLock pool remaining = do
locked <-
expectPoolUsage
=<< Pool.use pool (TxSessions.transactionNoRetry TxSessions.ReadCommitted TxSessions.Read (Tx.statement () groupRowShareLockStmt))
if locked
then pure ()
else threadDelay 20_000 >> waitForGroupRowLock pool (remaining - 1)
waitForPromotionGroupLock :: Pool.Pool -> Int -> IO ()
waitForPromotionGroupLock _ 0 = expectationFailure "promotion did not acquire its group lock in time"
waitForPromotionGroupLock pool remaining = do
locked <-
expectPoolUsage
=<< Pool.use pool (Session.statement () promotionGroupLockStmt)
if locked
then pure ()
else threadDelay 20_000 >> waitForPromotionGroupLock pool (remaining - 1)
waitForNamedSession :: Pool.Pool -> Text -> Int -> IO ()
waitForNamedSession _ _ 0 = expectationFailure "named session did not acquire the group share lock in time"
waitForNamedSession pool applicationName remaining = do
locked <- expectPoolUsage =<< Pool.use pool (Session.statement applicationName namedSessionGroupShareStmt)
if locked
then pure ()
else threadDelay 20_000 >> waitForNamedSession pool applicationName (remaining - 1)
waitForApplicationPid :: Pool.Pool -> Text -> Int -> IO Int32
waitForApplicationPid _ _ 0 = expectationFailure "named repair session did not become active in time" >> error "unreachable"
waitForApplicationPid pool applicationName remaining = do
maybePid <- expectPoolUsage =<< Pool.use pool (Session.statement applicationName applicationPidStmt)
case maybePid of
Just pid -> pure pid
Nothing -> threadDelay 20_000 >> waitForApplicationPid pool applicationName (remaining - 1)
expectStore ::
Store.KirokuStore ->
Eff '[Store, Error StoreError, IOE] value ->
IO value
expectStore store action =
Store.runStoreIO store action >>= \case
Left err -> expectationFailure (show err) >> error "unreachable"
Right value -> pure value
requireRight :: (Show error) => Either error value -> IO value
requireRight = \case
Left err -> expectationFailure (show err) >> error "unreachable"
Right value -> pure value
hasSqlState :: (Show error) => String -> Either error value -> Bool
hasSqlState wanted = \case
Left err -> wanted `List.isInfixOf` show err
Right _ -> False
requireIdentity :: (Show error) => Either error value -> value
requireIdentity = either (error . show) id
identity :: (Text -> Either error value) -> Text -> value
identity constructor = either (error . const "invalid test identity") id . constructor
run :: Text -> RebuildRunId
run = either (error . Text.unpack) id . mkRebuildRunId
driveVersionedToPromotion ::
Store.KirokuStore ->
ValidatedProjectionCatalog ->
RebuildRunId ->
Int ->
IO VersionedRebuildReport
driveVersionedToPromotion store catalog runId attempts
| attempts <= 0 = expectationFailure "versioned rebuild did not promote" >> error "unreachable"
| otherwise = do
report <- expectStore store (resumeVersionedRebuild catalog runId) >>= requireRight
if report ^. #phase == VersionedPromoted
then pure report
else driveVersionedToPromotion store catalog runId (attempts - 1)
driveVersionedToCutoverReady ::
Store.KirokuStore ->
ValidatedProjectionCatalog ->
RebuildRunId ->
Int ->
IO VersionedRebuildReport
driveVersionedToCutoverReady store catalog runId attempts
| attempts <= 0 = expectationFailure "versioned rebuild did not reach cutover" >> error "unreachable"
| otherwise = do
report <- expectStore store (resumeVersionedRebuild catalog runId) >>= requireRight
if report ^. #phase == VersionedCutoverReplaying && allVersionedComplete (report ^. #sources)
then pure report
else driveVersionedToCutoverReady store catalog runId (attempts - 1)
driveVersionedReplayComplete ::
Store.KirokuStore ->
ValidatedProjectionCatalog ->
RebuildRunId ->
Int ->
IO VersionedRebuildReport
driveVersionedReplayComplete store catalog runId attempts
| attempts <= 0 = expectationFailure "versioned rebuild did not complete ordinary replay" >> error "unreachable"
| otherwise = do
report <- expectStore store (resumeVersionedRebuild catalog runId) >>= requireRight
if report ^. #phase == VersionedReplayRunning && allVersionedComplete (report ^. #sources)
then pure report
else driveVersionedReplayComplete store catalog runId (attempts - 1)
assertPreparedV1Authority :: Store.KirokuStore -> RebuildRunId -> Bool -> IO ()
assertPreparedV1Authority store runId expectedPrepared = do
report <- expectStore store (inspectVersionedRebuild runId) >>= requireRight
report ^. #servingRevisionId `shouldBe` identity mkProjectionRevisionId "counter-v1"
report ^. #servingEpoch `shouldBe` 0
report ^. #promotionPrepared `shouldBe` expectedPrepared
status <- expectStore store (lookupProjectionGroupStatus mainGroupId)
status ^? _Just . #lifecyclePhase `shouldBe` Just "cutover-versioned"
status ^? _Just . #servingRevisionId
`shouldBe` Just (Just (identity mkProjectionRevisionId "counter-v1"))
status ^? _Just . #servingEpoch `shouldBe` Just 0
runStatement store () externalV1RowsStmt `shouldReturn` [(100, 42)]
runStatement store () servingCountsStmt `shouldReturn` (1, 1)
assertCandidateCounts :: Store.KirokuStore -> VersionedRebuildHandle -> Int64 -> IO ()
assertCandidateCounts store handle expected =
for_ (handle ^. #candidateGenerations) $ \generation ->
rowCount store (generation ^. #physicalTable) `shouldReturn` expected
allVersionedComplete :: [VersionedSourceProgress] -> Bool
allVersionedComplete =
all (\source -> source ^. #exhaustedThrough == Just (source ^. #targetPosition))
appendVersionedEvents :: Store.KirokuStore -> Text -> Int -> IO ()
appendVersionedEvents store streamName count = do
appended <-
Store.runStoreIO store $
Store.appendToStream
(StreamName streamName)
NoStream
[ EventData
{ eventId = Nothing,
eventType = EventType "VersionedDispatch",
payload = Aeson.Null,
metadata = Nothing,
causationId = Nothing,
correlationId = Nothing
}
| _ <- [1 .. count]
]
appended `shouldSatisfy` isRight
appendRepairEvents :: Store.KirokuStore -> Text -> Int64 -> [Int64] -> IO ()
appendRepairEvents store streamName aggregateId deltas = do
appended <-
Store.runStoreIO store $
Store.appendToStream
(StreamName streamName)
NoStream
[ EventData
{ eventId = Nothing,
eventType = EventType "RepairDelta",
payload = Aeson.toJSON [aggregateId, delta],
metadata = Nothing,
causationId = Nothing,
correlationId = Nothing
}
| delta <- deltas
]
appended `shouldSatisfy` isRight
appendMalformedRepairEvent :: Store.KirokuStore -> Text -> IO ()
appendMalformedRepairEvent store streamName = do
appended <-
Store.runStoreIO store $
Store.appendToStream
(StreamName streamName)
AnyVersion
[ EventData
{ eventId = Nothing,
eventType = EventType "RepairDelta",
payload = Aeson.object ["malformed" Aeson..= True],
metadata = Nothing,
causationId = Nothing,
correlationId = Nothing
}
]
appended `shouldSatisfy` isRight
deleteRepairRowStmt :: QualifiedTable -> Statement Int64 Int64
deleteRepairRowStmt table =
preparable
( "DELETE FROM "
<> qualifyTable (table ^. #schemaName) (table ^. #tableName)
<> " WHERE id = $1"
)
(E.param (E.nonNullable E.int8))
D.rowsAffected
upsertRepairRowStmt :: QualifiedTable -> Statement (Int64, Int64, Int64) ()
upsertRepairRowStmt table =
preparable
( "INSERT INTO "
<> qualifyTable (table ^. #schemaName) (table ^. #tableName)
<> " AS current_row"
<> " (id, detail, source_position) VALUES ($1, $2::text, $3) "
<> "ON CONFLICT (id) DO UPDATE SET "
<> "detail = (current_row.detail::bigint + $2)::text, source_position = $3"
)
( contrazip3
(E.param (E.nonNullable E.int8))
(E.param (E.nonNullable E.int8))
(E.param (E.nonNullable E.int8))
)
D.noResult
countRepairRowsStmt :: QualifiedTable -> Statement Int64 Int64
countRepairRowsStmt table =
preparable
( "SELECT count(*) FROM "
<> qualifyTable (table ^. #schemaName) (table ^. #tableName)
<> " WHERE id = $1"
)
(E.param (E.nonNullable E.int8))
(D.singleRow (D.column (D.nonNullable D.int8)))
seedRepairRowsStmt :: Statement () ()
seedRepairRowsStmt =
preparable
"""
INSERT INTO app.counter_audit (id, detail, source_position)
VALUES (1, '999', 99), (2, '7', 1)
"""
E.noParams
D.noResult
seedSingleRepairRowStmt :: Statement () ()
seedSingleRepairRowStmt =
preparable
"""
INSERT INTO app.counter_audit (id, detail, source_position)
VALUES (1, '999', 99)
"""
E.noParams
D.noResult
repairRowsStmt :: Statement () [(Int64, Text)]
repairRowsStmt =
preparable
"SELECT id, detail FROM app.counter_audit ORDER BY id"
E.noParams
( D.rowList
( (,)
<$> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.text)
)
)
sanctionedRepairReaderLockStmt :: Statement Text Text
sanctionedRepairReaderLockStmt =
preparable
"SELECT group_id FROM keiro.keiro_projection_rebuild_groups WHERE group_id = $1 AND reads_allowed FOR SHARE"
(E.param (E.nonNullable E.text))
(D.singleRow (D.column (D.nonNullable D.text)))
repairDetailByIdStmt :: Statement Int64 Text
repairDetailByIdStmt =
preparable
"SELECT detail FROM app.counter_audit WHERE id = $1"
(E.param (E.nonNullable E.int8))
(D.singleRow (D.column (D.nonNullable D.text)))
repairDedupCountStmt :: Statement () Int64
repairDedupCountStmt =
preparable
"SELECT count(*) FROM keiro.keiro_projection_dedup WHERE projection_name = 'catalog-async'"
E.noParams
(D.singleRow (D.column (D.nonNullable D.int8)))
repairCheckpointStmt :: Statement () (Maybe Int64)
repairCheckpointStmt =
preparable
"""
SELECT checkpoint_position
FROM kiroku.subscription_checkpoints_v1
WHERE subscription_name = 'catalog-async-subscription'
"""
E.noParams
(D.rowMaybe (D.column (D.nonNullable D.int8)))
forceGroupSliceStmt :: Statement Text ()
forceGroupSliceStmt =
preparable
"UPDATE keiro.keiro_projection_rebuild_groups SET slice_fingerprint = $1 WHERE group_id = 'counter-group'"
(E.param (E.nonNullable E.text))
D.noResult
relationOidFor :: Store.KirokuStore -> QualifiedTable -> IO Int64
relationOidFor store table = do
result <- runStatement store table relationOidStmt
maybe (expectationFailure "relation missing" >> error "unreachable") pure result
relationOidStmt :: Statement QualifiedTable (Maybe Int64)
relationOidStmt =
preparable
"""
SELECT classes.oid::bigint
FROM pg_catalog.pg_class AS classes
JOIN pg_catalog.pg_namespace AS namespaces
ON namespaces.oid = classes.relnamespace
WHERE namespaces.nspname = $1 AND classes.relname = $2
"""
( (\table -> (table ^. #schemaName, table ^. #tableName))
>$< contrazip2
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
)
(D.rowMaybe (D.column (D.nonNullable D.int8)))
relationExistsStmt :: Statement QualifiedTable Bool
relationExistsStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_class AS classes
JOIN pg_catalog.pg_namespace AS namespaces
ON namespaces.oid = classes.relnamespace
WHERE namespaces.nspname = $1 AND classes.relname = $2
)
"""
( (\table -> (table ^. #schemaName, table ^. #tableName))
>$< contrazip2
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
)
(D.singleRow (D.column (D.nonNullable D.bool)))
rogueColumnStmt :: Statement QualifiedTable Bool
rogueColumnStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = $1 AND table_name = $2 AND column_name = 'rogue'
)
"""
( (\table -> (table ^. #schemaName, table ^. #tableName))
>$< contrazip2
(E.param (E.nonNullable E.text))
(E.param (E.nonNullable E.text))
)
(D.singleRow (D.column (D.nonNullable D.bool)))
rowCount :: Store.KirokuStore -> QualifiedTable -> IO Int64
rowCount store table =
runStatement store () $
preparable
( "SELECT count(*) FROM "
<> qualifyTable (table ^. #schemaName) (table ^. #tableName)
)
E.noParams
(D.singleRow (D.column (D.nonNullable D.int8)))
servingCountsStmt :: Statement () (Int64, Int64)
servingCountsStmt =
preparable
"""
SELECT (SELECT count(*) FROM app.counter),
(SELECT count(*) FROM app.counter_audit)
"""
E.noParams
( D.singleRow
( (,)
<$> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.int8)
)
)
counterCountStmt :: Statement () Int64
counterCountStmt =
preparable
"SELECT count(*) FROM app.counter"
E.noParams
(D.singleRow (D.column (D.nonNullable D.int8)))
auditCountStmt :: Statement () Int64
auditCountStmt =
preparable
"SELECT count(*) FROM app.counter_audit"
E.noParams
(D.singleRow (D.column (D.nonNullable D.int8)))
externalV1RowsStmt :: Statement () [(Int64, Int64)]
externalV1RowsStmt =
preparable
"SELECT id, total FROM keiro_read.counter_reader_v1() ORDER BY id"
E.noParams
( D.rowList
( (,)
<$> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.int8)
)
)
externalV2RowsStmt :: Statement () [(Int64, Int64, Int64, Int64)]
externalV2RowsStmt =
preparable
"SELECT id, subtotal, tax, total FROM keiro_read.counter_reader_v2() ORDER BY id"
E.noParams
( D.rowList
( (,,,)
<$> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.int8)
<*> D.column (D.nonNullable D.int8)
)
)
counterAccessShareStmt :: Statement () Bool
counterAccessShareStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM pg_locks
WHERE relation = 'app.counter'::regclass
AND mode = 'AccessShareLock'
AND granted
AND pid <> pg_backend_pid()
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
auditAccessShareStmt :: Statement () Bool
auditAccessShareStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM pg_locks
WHERE relation = 'app.counter_audit'::regclass
AND mode = 'AccessShareLock'
AND granted
AND pid <> pg_backend_pid()
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
lockGroupRowStmt :: Statement Text Text
lockGroupRowStmt =
preparable
"UPDATE keiro.keiro_projection_rebuild_groups SET updated_at = updated_at WHERE group_id = $1 RETURNING group_id"
(E.param (E.nonNullable E.text))
(D.singleRow (D.column (D.nonNullable D.text)))
groupRowShareLockStmt :: Statement () Bool
groupRowShareLockStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM pg_locks
WHERE relation = 'keiro.keiro_projection_rebuild_groups'::regclass
AND mode = 'RowExclusiveLock'
AND granted
AND pid <> pg_backend_pid()
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
promotionGroupLockStmt :: Statement () Bool
promotionGroupLockStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_locks
WHERE relation = 'keiro.keiro_projection_rebuild_groups'::regclass
AND mode = 'RowShareLock'
AND granted
AND pid <> pg_backend_pid()
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
namedSessionGroupShareStmt :: Statement Text Bool
namedSessionGroupShareStmt =
preparable
"""
SELECT EXISTS (
SELECT 1 FROM pg_catalog.pg_stat_activity AS activity
JOIN pg_catalog.pg_locks AS locks
ON locks.pid = activity.pid
WHERE activity.application_name = $1
AND locks.relation = 'keiro.keiro_projection_rebuild_groups'::regclass
AND locks.mode = 'RowShareLock'
AND locks.granted
)
"""
(E.param (E.nonNullable E.text))
(D.singleRow (D.column (D.nonNullable D.bool)))
applicationPidStmt :: Statement Text (Maybe Int32)
applicationPidStmt =
preparable
"SELECT pid FROM pg_catalog.pg_stat_activity WHERE application_name = $1 AND state = 'active' ORDER BY pid LIMIT 1"
(E.param (E.nonNullable E.text))
(D.rowMaybe (D.column (D.nonNullable D.int4)))
terminateBackendStmt :: Statement Int32 Bool
terminateBackendStmt =
preparable
"SELECT pg_catalog.pg_terminate_backend($1)"
(E.param (E.nonNullable E.int4))
(D.singleRow (D.column (D.nonNullable D.bool)))
statusPromotionUpdateAndSleepSql :: ByteString
statusPromotionUpdateAndSleepSql =
"""
UPDATE keiro.keiro_projection_rebuild_groups
SET status = 'serving-versioned',
active_run_id = NULL,
serving_revision_id = 'counter-v2',
serving_epoch = 1,
completed_at = now(),
updated_at = now()
WHERE group_id = 'counter-group';
SELECT pg_sleep(1);
"""
dispatchDedupCountStmt :: Statement () Int64
dispatchDedupCountStmt =
preparable
"""
SELECT count(*)
FROM keiro.keiro_projection_dedup
WHERE projection_name = 'catalog-async'
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.int8)))
convergeDedupCountStmt :: Statement () Int64
convergeDedupCountStmt = dispatchDedupCountStmt
expiredRetentionFactsStmt :: Statement () (Text, Bool, Bool, Text, Text)
expiredRetentionFactsStmt =
preparable
"""
SELECT groups.status, groups.reads_allowed, groups.writes_allowed,
runs.status, runs.failure_code
FROM keiro.keiro_projection_rebuild_groups AS groups
JOIN keiro.keiro_projection_rebuild_runs AS runs
ON runs.group_id = groups.group_id
WHERE runs.run_id = 'versioned-expired-retention'
"""
E.noParams
( D.singleRow
( (,,,,)
<$> column D.text
<*> column D.bool
<*> column D.bool
<*> column D.text
<*> column D.text
)
)
where
column = D.column . D.nonNullable
upsertSubscriptionCursorStmt :: Statement (Text, Int64) ()
upsertSubscriptionCursorStmt =
preparable
"""
INSERT INTO subscriptions (subscription_name, stream_name, last_seen)
VALUES ($1, '$all', $2)
ON CONFLICT (subscription_name, consumer_group_member) DO UPDATE
SET last_seen = EXCLUDED.last_seen,
updated_at = now()
"""
(contrazip2 (E.param (E.nonNullable E.text)) (E.param (E.nonNullable E.int8)))
D.noResult
promotedLifecycleFactsStmt :: Statement () (Text, Bool, Bool, Text, Int64, Text, Int64, Int64, Int64)
promotedLifecycleFactsStmt =
preparable
"""
SELECT groups.status, groups.reads_allowed, groups.writes_allowed,
groups.serving_revision_id, groups.serving_epoch, runs.status,
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'serving'),
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'retired'),
(SELECT count(*) FROM kiroku.history_retention_leases WHERE released_at IS NOT NULL)
FROM keiro.keiro_projection_rebuild_groups AS groups
JOIN keiro.keiro_projection_rebuild_runs AS runs ON runs.group_id = groups.group_id
WHERE runs.run_id = 'versioned-promote'
"""
E.noParams
( D.singleRow
( (,,,,,,,,)
<$> column D.text
<*> column D.bool
<*> column D.bool
<*> column D.text
<*> column D.int8
<*> column D.text
<*> column D.int8
<*> column D.int8
<*> column D.int8
)
)
where
column = D.column . D.nonNullable
promotedCounterShapeStmt :: Statement () Bool
promotedCounterShapeStmt =
preparable
"""
SELECT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = 'app' AND table_name = 'counter'
AND column_name = 'subtotal'
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
cloneShapeStmt :: Statement () Bool
cloneShapeStmt =
preparable
"""
SELECT
(SELECT count(*) FROM information_schema.columns
WHERE table_schema = 'app' AND table_name = 'counter') = 2
AND EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = 'app' AND table_name = 'counter'
AND column_name = 'total'
)
AND NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = 'app' AND table_name = 'counter'
AND column_name = 'subtotal'
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
identityCloneObjectsStmt :: Statement () Bool
identityCloneObjectsStmt =
preparable
"""
SELECT
pg_catalog.pg_get_serial_sequence('app.counter', 'id') = 'app.counter_id_seq'
AND EXISTS (
SELECT 1
FROM pg_catalog.pg_constraint
WHERE conrelid = 'app.counter'::regclass
AND conname = 'counter_pkey'
AND contype = 'p'
)
"""
E.noParams
(D.singleRow (D.column (D.nonNullable D.bool)))
activeLifecycleFactsStmt :: Statement () (Text, Bool, Bool, Text, Int64, Int64, Int64, Int64, Int64)
activeLifecycleFactsStmt =
preparable
"""
SELECT groups.status, groups.reads_allowed, groups.writes_allowed,
groups.serving_revision_id, groups.serving_epoch,
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'serving'),
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'staging'),
(SELECT count(*) FROM keiro.keiro_projection_rebuild_runs WHERE rebuild_mode = 'versioned'),
(SELECT count(*) FROM kiroku.history_retention_leases WHERE released_at IS NULL)
FROM keiro.keiro_projection_rebuild_groups AS groups
WHERE groups.group_id = 'counter-group'
"""
E.noParams
( D.singleRow
( (,,,,,,,,)
<$> column D.text
<*> column D.bool
<*> column D.bool
<*> column D.text
<*> column D.int8
<*> column D.int8
<*> column D.int8
<*> column D.int8
<*> column D.int8
)
)
where
column = D.column . D.nonNullable
rolledBackLifecycleFactsStmt :: Statement () (Text, Int64, Int64, Int64, Int64)
rolledBackLifecycleFactsStmt =
preparable
"""
SELECT groups.status,
(SELECT count(*) FROM keiro.keiro_projection_target_generations),
(SELECT count(*) FROM keiro.keiro_projection_rebuild_runs WHERE rebuild_mode = 'versioned'),
(SELECT count(*) FROM kiroku.history_retention_leases),
(SELECT count(*) FROM pg_catalog.pg_class AS classes JOIN pg_catalog.pg_namespace AS namespaces ON namespaces.oid = classes.relnamespace WHERE namespaces.nspname = 'app' AND classes.relname LIKE 'keiro_g_%')
FROM keiro.keiro_projection_rebuild_groups AS groups
WHERE groups.group_id = 'counter-group'
"""
E.noParams
( D.singleRow
( (,,,,)
<$> column D.text
<*> column D.int8
<*> column D.int8
<*> column D.int8
<*> column D.int8
)
)
where
column = D.column . D.nonNullable
collisionRollbackFactsStmt :: Statement () (Text, Int64, Int64, Int64, Int64)
collisionRollbackFactsStmt = rolledBackLifecycleFactsStmt
abandonedLifecycleFactsStmt :: Statement () (Text, Bool, Bool, Text, Int64, Int64, Int64)
abandonedLifecycleFactsStmt =
preparable
"""
SELECT groups.status, groups.reads_allowed, groups.writes_allowed, runs.status,
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'dropped'),
(SELECT count(*) FROM kiroku.history_retention_leases WHERE released_at IS NOT NULL),
(SELECT count(*) FROM keiro.keiro_projection_target_generations WHERE lifecycle = 'serving')
FROM keiro.keiro_projection_rebuild_groups AS groups
JOIN keiro.keiro_projection_rebuild_runs AS runs
ON runs.group_id = groups.group_id
WHERE runs.run_id = 'versioned-abandon'
"""
E.noParams
( D.singleRow
( (,,,,,,)
<$> column D.text
<*> column D.bool
<*> column D.bool
<*> column D.text
<*> column D.int8
<*> column D.int8
<*> column D.int8
)
)
where
column = D.column . D.nonNullable
isLeft :: Either a b -> Bool
isLeft = \case
Left _ -> True
Right _ -> False
isRight :: Either a b -> Bool
isRight = not . isLeft