packages feed

clickhouse-migrate (empty) → 0.1.0.0

raw patch · 66 files changed

+5355/−0 lines, 66 filesdep +aesondep +basedep +bytestring

Dependencies added: aeson, base, bytestring, containers, cryptohash-sha256, entropy, http-client, http-client-tls, http-types, text, time

Files

+ CHANGELOG.md view
@@ -0,0 +1,32 @@+# Changelog for clickhouse-migrate++## 0.1.0.0 — 2026-10-16++Initial release.++### New Features++- Validated, ordered migration definitions and plans built through public smart+  constructors, with exact SQL byte preservation and a constrained statement+  envelope (`ClickHouseMigrate.Definition`, `ClickHouseMigrate.Plan`).+- Checksum v1: context-aware SHA-256 over immutable plan views, verified against+  independent test vectors.+- Explicit component catalogs separated from an explicit execution schedule, so+  components can be upgraded under a declared migration order.+- Conservative direct HTTP transport to ClickHouse with validated, application-owned+  `TargetConfig` and no implicit credentials (`ClickHouseMigrate.Target`).+- Immutable ledger events with guarded bootstrap, qualified uncertain appends,+  guarded reconciliation and complete-history reads with bounded row decoding+  (`ClickHouseMigrate.Ledger`, `ClickHouseMigrate.Event`).+- Externally serialized exclusive administrative access+  (`ClickHouseMigrate.ExclusiveAccess`).+- Execution (`up`), status, history and verify with scoped evidence; uncertain+  work blocks ordinary `up` (`ClickHouseMigrate.Runner`,+  `ClickHouseMigrate.Inspection`, `ClickHouseMigrate.Report`).+- Evidence-guarded repair and recovery that validate complete recovery events+  before any mutation (`ClickHouseMigrate.Repair`).+- Source-neutral legacy history adoption preserving reviewed source evidence and+  native baselines (`ClickHouseMigrate.Adoption`).+- Asynchronous cancellation propagates after bounded cleanup.++Qualified against ClickHouse 26.8.28.14 LTS (official Apple Silicon macOS binary).
+ LICENSE view
@@ -0,0 +1,25 @@+Copyright (c) 2026, shinzui+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+   list of conditions and the following disclaimer.+2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.+3. Neither the name of the copyright holder nor the names of its contributors+   may be used to endorse or promote products derived from this software without+   specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,22 @@+# clickhouse-migrate++Validated ordered migration definitions, exact-byte checksum v1, conservative direct+HTTP execution, immutable ledger history, scoped inspection/recovery and source-neutral+legacy adoption. No runtime file discovery, implicit credentials, printing or process+exit. Optional embedding, CLI and server fixtures are separate packages.++Construct `MigrationPlan` with the public smart constructors; supply validated+`TargetConfig`, a mandatory externally serialized administrative provider, event sink+and diagnostic policy to `mutationContext`. `up` initializes only with a retained+explicit target/UUID authorization. Uncertain work blocks ordinary up. Status/history+never initialize; verify checks definition compatibility, optionally currency, not+schema equality. All repair/adoption operations require explicit evidence and fresh+expected state. Async cancellation propagates after bounded cleanup.++Qualified target: exact ClickHouse 26.8.28.14 LTS official Apple Silicon macOS binary,+direct self-managed Atomic/MergeTree on healthy writable Local storage. No Cloud,+proxy, distributed/replicated, Linux or power-loss claim. Runtime version probes do+not attest platform/digest. The workspace's docs/user/contracts and docs/user/operations.md+publish frozen formats, server artifact identity and coordination/recovery limits.+The standard Haskell process library is a transitive HTTP dependency; fixture tools,+optional packages and the opt-in internal driver are absent from production outputs.
+ clickhouse-migrate.cabal view
@@ -0,0 +1,205 @@+cabal-version: 3.12+name: clickhouse-migrate+version: 0.1.0.0+synopsis: Core definitions and execution for ClickHouse migrations+description:+  Core definitions and execution for ClickHouse migrations.+  Validated public APIs with explicit application-owned configuration and scope.++license: BSD-3-Clause+license-file: LICENSE+author: shinzui+maintainer: shinzui@users.noreply.github.com+category: Database+build-type: Simple+extra-doc-files:+  CHANGELOG.md+  README.md++extra-source-files:+  test/public/*.hs+  test/vectors/*.json+  test/vectors/*.tsv++tested-with: ghc ==9.12.4++source-repository head+  type: git+  location: https://github.com/shinzui/clickhouse-migrate.git+  subdir: clickhouse-migrate++common common+  default-language: GHC2024+  default-extensions:+    DeriveAnyClass+    DuplicateRecordFields+    OverloadedLabels+    OverloadedStrings++  ghc-options: -Wall+  build-depends:+    base >=4.21 && <4.22,+    bytestring >=0.12.2 && <0.13,+    containers >=0.7 && <0.9,+    cryptohash-sha256 >=0.11.102.1 && <0.12,+    entropy >=0.4.1.10 && <0.5,+    text >=2.1 && <2.2,+    time >=1.12.2 && <1.17,++  build-depends:+    aeson >=2.2.4.1 && <2.4,+    http-client ==0.7.19,+    http-client-tls >=0.3.6.4 && <0.5,+    http-types >=0.12.4 && <0.13,++library+  import: common+  hs-source-dirs: src+  exposed-modules:+    ClickHouseMigrate+    ClickHouseMigrate.Adoption+    ClickHouseMigrate.Definition+    ClickHouseMigrate.Error+    ClickHouseMigrate.Event+    ClickHouseMigrate.ExclusiveAccess+    ClickHouseMigrate.Inspection+    ClickHouseMigrate.Ledger+    ClickHouseMigrate.Plan+    ClickHouseMigrate.Prelude+    ClickHouseMigrate.Repair+    ClickHouseMigrate.Report+    ClickHouseMigrate.Runner+    ClickHouseMigrate.Target++  other-modules:+    ClickHouseMigrate.Internal.AdoptionTypes+    ClickHouseMigrate.Internal.AdoptionValidation+    ClickHouseMigrate.Internal.Checksum+    ClickHouseMigrate.Internal.EventTypes+    ClickHouseMigrate.Internal.Exclusive+    ClickHouseMigrate.Internal.ExclusiveTypes+    ClickHouseMigrate.Internal.Execution+    ClickHouseMigrate.Internal.Http+    ClickHouseMigrate.Internal.Identity+    ClickHouseMigrate.Internal.LedgerCodec+    ClickHouseMigrate.Internal.LedgerSchema+    ClickHouseMigrate.Internal.LedgerStore+    ClickHouseMigrate.Internal.LedgerTypes+    ClickHouseMigrate.Internal.LedgerValidation+    ClickHouseMigrate.Internal.Preflight+    ClickHouseMigrate.Internal.Projection+    ClickHouseMigrate.Internal.RepairTypes+    ClickHouseMigrate.Internal.Reporting+    ClickHouseMigrate.Internal.RuntimeTypes+    ClickHouseMigrate.Internal.ServerPolicy+    ClickHouseMigrate.Internal.SqlEnvelope+    ClickHouseMigrate.Internal.TargetTypes+    ClickHouseMigrate.Internal.Transport+    ClickHouseMigrate.Internal.Types++flag internal-test-driver+  description: Build private runtime qualification driver (supplied by EP-4)+  default: False+  manual: True++executable clickhouse-migrate-test-driver+  import: common+  ghc-options: -threaded+  hs-source-dirs:+    test-driver+    src++  main-is: Main.hs+  other-modules:+    ClickHouseMigrate+    ClickHouseMigrate.Adoption+    ClickHouseMigrate.Definition+    ClickHouseMigrate.Error+    ClickHouseMigrate.Event+    ClickHouseMigrate.ExclusiveAccess+    ClickHouseMigrate.Inspection+    ClickHouseMigrate.Internal.AdoptionTypes+    ClickHouseMigrate.Internal.AdoptionValidation+    ClickHouseMigrate.Internal.Checksum+    ClickHouseMigrate.Internal.EventTypes+    ClickHouseMigrate.Internal.Exclusive+    ClickHouseMigrate.Internal.ExclusiveTypes+    ClickHouseMigrate.Internal.Execution+    ClickHouseMigrate.Internal.Http+    ClickHouseMigrate.Internal.Identity+    ClickHouseMigrate.Internal.LedgerCodec+    ClickHouseMigrate.Internal.LedgerSchema+    ClickHouseMigrate.Internal.LedgerStore+    ClickHouseMigrate.Internal.LedgerTypes+    ClickHouseMigrate.Internal.LedgerValidation+    ClickHouseMigrate.Internal.Preflight+    ClickHouseMigrate.Internal.Projection+    ClickHouseMigrate.Internal.RepairTypes+    ClickHouseMigrate.Internal.Reporting+    ClickHouseMigrate.Internal.RuntimeTypes+    ClickHouseMigrate.Internal.ServerPolicy+    ClickHouseMigrate.Internal.SqlEnvelope+    ClickHouseMigrate.Internal.TargetTypes+    ClickHouseMigrate.Internal.Transport+    ClickHouseMigrate.Internal.Types+    ClickHouseMigrate.Ledger+    ClickHouseMigrate.Plan+    ClickHouseMigrate.Prelude+    ClickHouseMigrate.Repair+    ClickHouseMigrate.Report+    ClickHouseMigrate.Runner+    ClickHouseMigrate.Target++  if !flag(internal-test-driver)+    buildable: False++test-suite clickhouse-migrate-unit+  import: common+  type: exitcode-stdio-1.0+  hs-source-dirs:+    test+    src++  main-is: Main.hs+  other-modules:+    AdoptionTests+    ClickHouseMigrate+    ClickHouseMigrate.Adoption+    ClickHouseMigrate.Definition+    ClickHouseMigrate.Error+    ClickHouseMigrate.Event+    ClickHouseMigrate.ExclusiveAccess+    ClickHouseMigrate.Inspection+    ClickHouseMigrate.Internal.AdoptionTypes+    ClickHouseMigrate.Internal.AdoptionValidation+    ClickHouseMigrate.Internal.Checksum+    ClickHouseMigrate.Internal.EventTypes+    ClickHouseMigrate.Internal.Exclusive+    ClickHouseMigrate.Internal.ExclusiveTypes+    ClickHouseMigrate.Internal.Execution+    ClickHouseMigrate.Internal.Http+    ClickHouseMigrate.Internal.Identity+    ClickHouseMigrate.Internal.LedgerCodec+    ClickHouseMigrate.Internal.LedgerSchema+    ClickHouseMigrate.Internal.LedgerStore+    ClickHouseMigrate.Internal.LedgerTypes+    ClickHouseMigrate.Internal.LedgerValidation+    ClickHouseMigrate.Internal.Preflight+    ClickHouseMigrate.Internal.Projection+    ClickHouseMigrate.Internal.RepairTypes+    ClickHouseMigrate.Internal.Reporting+    ClickHouseMigrate.Internal.RuntimeTypes+    ClickHouseMigrate.Internal.ServerPolicy+    ClickHouseMigrate.Internal.SqlEnvelope+    ClickHouseMigrate.Internal.TargetTypes+    ClickHouseMigrate.Internal.Transport+    ClickHouseMigrate.Internal.Types+    ClickHouseMigrate.Ledger+    ClickHouseMigrate.Plan+    ClickHouseMigrate.Prelude+    ClickHouseMigrate.Repair+    ClickHouseMigrate.Report+    ClickHouseMigrate.Runner+    ClickHouseMigrate.Target+    ExclusiveTests+    LedgerTests
+ src/ClickHouseMigrate.hs view
@@ -0,0 +1,52 @@+-- | Ordered definitions, scoped execution, inspection, recovery and adoption.+module ClickHouseMigrate+  ( module ClickHouseMigrate.Adoption,+    module ClickHouseMigrate.Definition,+    module ClickHouseMigrate.Plan,+    module ClickHouseMigrate.Error,+    module ClickHouseMigrate.Target,+    MutationContext,+    mutationContext,+    UpOptions,+    upOptions,+    up,+    InspectionOptions (Observe, Guarded),+    VerifyOptions,+    verifyOptions,+    status,+    history,+    verify,+    RepairRequest,+    RecoveryEvidence,+    RepairRequestError (..),+    recoveryEvidence,+    resumeRequest,+    retryStepRequest,+    acceptStepRequest,+    finalizeRequest,+    repair,+    EventSink,+    DiagnosticPolicy,+    quietEvents,+    redactedDiagnostics,+    sqlDiagnostics,+    RunReport,+    RunFailure,+    InspectionReport,+    InspectionFailure,+    HistoryReport,+    VerificationReport,+  )+where++import ClickHouseMigrate.Adoption+import ClickHouseMigrate.Definition+import ClickHouseMigrate.Error+import ClickHouseMigrate.Plan+import ClickHouseMigrate.Target++import ClickHouseMigrate.Event (DiagnosticPolicy, EventSink, quietEvents, redactedDiagnostics, sqlDiagnostics)+import ClickHouseMigrate.Inspection+import ClickHouseMigrate.Repair+import ClickHouseMigrate.Report (HistoryReport, InspectionFailure, InspectionReport, RunFailure, RunReport, VerificationReport)+import ClickHouseMigrate.Runner
+ src/ClickHouseMigrate/Adoption.hs view
@@ -0,0 +1,102 @@+-- | Reviewed source-neutral adoption into fresh history. Adoption never runs SQL.+module ClickHouseMigrate.Adoption+  ( AdoptionError (..),+    SourceChecksum,+    sourceChecksum,+    LegacyRow,+    legacyRow,+    AdoptionMapping,+    attestedMapping,+    verifiedMapping,+    AdoptionRequest,+    adoptionRequest,+    AdoptionPlan,+    planAdoption,+    inspectAdoption,+    adoptionEntries,+    applyAdoption,+    resumeAdoption,+  )+where++import ClickHouseMigrate.Event qualified as Event+import ClickHouseMigrate.Internal.AdoptionTypes+import ClickHouseMigrate.Internal.AdoptionValidation (adoptionRequest, legacyRow, sourceChecksum)+import ClickHouseMigrate.Internal.AdoptionValidation qualified as Validation+import ClickHouseMigrate.Internal.Execution+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Internal.RuntimeTypes (MutationContext (..))+import ClickHouseMigrate.Internal.TargetTypes qualified as T+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R++-- | Filename/time history, even with today's source bytes, remains attested.+attestedMapping :: Text -> Text -> Text -> Text -> [Text] -> Either AdoptionError AdoptionMapping+attestedMapping row component migration checksum refs = Validation.mapping row component migration checksum (Attested refs)++-- | Use only when the predecessor actually stored a supported checksum.+-- Supplied historical source bytes must match that stored value; this does not+-- prove that a partially executed legacy file completed or that native SQL ran.+verifiedMapping :: Text -> Text -> Text -> Text -> ByteString -> [Text] -> Either AdoptionError AdoptionMapping+verifiedMapping row component migration checksum bytes refs = Validation.mapping row component migration checksum (Verified bytes refs)++planAdoption :: Plan.MigrationPlan -> R.InspectionReport -> AdoptionRequest -> Either AdoptionError AdoptionPlan+planAdoption plan (R.InspectionReport snapshot _ _) request = do+  (batch, entries) <- Validation.buildBatch plan request+  _ <- Validation.checkHistory request batch entries snapshot+  pure (AdoptionPlan request batch entries)++inspectAdoption :: AdoptionPlan -> R.AdoptionSummary+inspectAdoption (AdoptionPlan _ (L.AdoptionBatch source snapshot mapping definitions count) _) = R.AdoptionSummary source snapshot mapping definitions count++adoptionEntries :: AdoptionPlan -> [(R.DefinitionSummary, R.EvidenceSummary)]+adoptionEntries (AdoptionPlan _ _ entries) = [(Reporting.summaryDefinition definition, R.EvidenceSummary level refs) | (definition, L.Evidence level refs) <- entries]++applyAdoption :: MutationContext -> Plan.MigrationPlan -> AdoptionPlan -> IO (Either R.RunFailure R.RunReport)+applyAdoption = executeAdoption False++resumeAdoption :: MutationContext -> Plan.MigrationPlan -> AdoptionPlan -> IO (Either R.RunFailure R.RunReport)+resumeAdoption = executeAdoption True++executeAdoption :: Bool -> MutationContext -> Plan.MigrationPlan -> AdoptionPlan -> IO (Either R.RunFailure R.RunReport)+executeAdoption resume context@(MutationContext (T.TargetConfig actual _ _ _ binding _ _) _ _ _) plan (AdoptionPlan request@(AdoptionRequest target _ authorization quiet audit _ _ _ _) batch entries) =+  let validate snapshot = either (Left . R.AdoptionRejected) Right $ do+        (current, definitions) <- Validation.buildBatch plan request+        Validation.require MappingDefinitionMismatch (current == batch && definitions == entries)+        committed <- Validation.checkHistory request batch entries snapshot+        unless committed $ do+          let R.HistoryReport _ _ _ _ _ events _ _ = snapshot+              interrupted = any (\case R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.AdoptionBeginning _) -> True; _ -> False) events+          Validation.require (if resume then AdoptionBatchConflict else AdoptionResumeRequired) (resume == interrupted)+        unless committed $ case snapshot of+          R.HistoryReport _ _ presence _ _ _ _ _ | presence `elem` [R.LedgerMissing, R.LedgerUninitialized] -> Validation.require AdoptionInitializationRequired (binding == T.FirstUse && isJust authorization)+          _ -> pure ()+        pure committed+      prerequisite = if target == actual then Right () else Left (R.AdoptionRejected AdoptionTargetMismatch)+   in withRuntimeObservation prerequisite (Just validate) context audit authorization (Just quiet) plan $ \runtime -> do+        projection <- runtimeProjection runtime+        case L.adoption projection of+          Nothing -> do+            when resume (abort (R.AdoptionRejected AdoptionBatchConflict))+            emit runtime Event.BeforeAdoptionStart Nothing Nothing Nothing+            appendEvent runtime Nothing (L.StartAdoption batch) Nothing+            emit runtime Event.AdoptionStarted Nothing Nothing Nothing+          Just (L.ActiveAdoption current _) -> do+            unless resume (abort (R.AdoptionRejected AdoptionResumeRequired))+            unless (current == batch) (abort (R.AdoptionRejected AdoptionBatchConflict))+        after <- runtimeProjection runtime+        completed <- case L.adoption after of Just (L.ActiveAdoption _ values) -> pure (length values); _ -> abort R.UnexpectedRuntimeFailure+        mapM_+          ( \(definition, evidence) -> do+              emit runtime Event.BeforeAdoptionEntry (Just definition) Nothing Nothing+              bound <- contextAt definition Nothing+              appendEvent runtime Nothing (L.AdoptEntry definition evidence) bound+              emit runtime Event.AdoptionEntryRecorded (Just definition) Nothing Nothing+          )+          (drop completed entries)+        let L.AdoptionBatch _ _ mapping _ _ = batch+        emit runtime Event.BeforeAdoptionComplete Nothing Nothing Nothing+        appendEvent runtime Nothing (L.CompleteAdoption mapping) Nothing+        emit runtime Event.AdoptionCompleted Nothing Nothing Nothing
+ src/ClickHouseMigrate/Definition.hs view
@@ -0,0 +1,108 @@+-- | Pure authoring values. Constructors and representations remain private.+module ClickHouseMigrate.Definition+  ( ComponentName,+    MigrationName,+    StepName,+    MigrationId,+    SqlStep,+    Migration,+    MigrationComponent,+    componentName,+    migrationName,+    stepName,+    componentNameText,+    migrationNameText,+    stepNameText,+    migrationIdFromNames,+    migrationIdComponent,+    migrationIdName,+    sqlStep,+    sqlMigration,+    migrationComponent,+    withMigrationDescription,+    sqlMigrationFromList,+    migrationComponentFromList,+  )+where++import ClickHouseMigrate.Error+import ClickHouseMigrate.Internal.SqlEnvelope (validateSql)+import ClickHouseMigrate.Internal.Types+import ClickHouseMigrate.Prelude+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text qualified as Text++validateName :: Text -> Text -> Either DefinitionError Text+validateName scope value+  | Text.length value <= 128 && maybe False (\(c, rest) -> first c && Text.all later rest) (Text.uncons value) = Right value+  | otherwise = Left (DefinitionError InvalidName [scope] Nothing)+  where+    first c = (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9')+    later c = first c || c == '_' || c == '-'++componentName :: Text -> Either DefinitionError ComponentName+componentName = fmap ComponentName . validateName "component"++migrationName :: Text -> Either DefinitionError MigrationName+migrationName = fmap MigrationName . validateName "migration"++stepName :: Text -> Either DefinitionError StepName+stepName = fmap StepName . validateName "step"++componentNameText :: ComponentName -> Text+componentNameText (ComponentName value) = value++migrationNameText :: MigrationName -> Text+migrationNameText (MigrationName value) = value++stepNameText :: StepName -> Text+stepNameText (StepName value) = value++-- | A schedule identity from already validated local names.+migrationIdFromNames :: ComponentName -> MigrationName -> MigrationId+migrationIdFromNames = MigrationId++migrationIdComponent :: MigrationId -> ComponentName+migrationIdComponent (MigrationId component _) = component++migrationIdName :: MigrationId -> MigrationName+migrationIdName (MigrationId _ name) = name++sqlStep :: Text -> ByteString -> Either DefinitionError SqlStep+sqlStep name sql = do+  validated <- stepName name+  case validateSql sql of+    Left (DefinitionError c path offset) -> Left (DefinitionError c (["step", name] ++ path) offset)+    Right () -> Right (SqlStep validated sql)++sqlMigration :: Text -> NonEmpty SqlStep -> Either DefinitionError Migration+sqlMigration name steps = do+  validated <- migrationName name+  unique ["migration", name, "steps"] [step | SqlStep step _ <- NE.toList steps]+  pure (Migration validated Nothing steps)++-- | Parsing boundary for formats which can represent empty lists.+sqlMigrationFromList :: Text -> [SqlStep] -> Either DefinitionError Migration+sqlMigrationFromList name steps = nonEmpty ["migration", name, "steps"] steps >>= sqlMigration name++withMigrationDescription :: Maybe Text -> Migration -> Migration+withMigrationDescription description (Migration name _ steps) = Migration name description steps++migrationComponent :: Text -> Set Text -> NonEmpty Migration -> Either DefinitionError MigrationComponent+migrationComponent name dependencies migrations = do+  validated <- componentName name+  deps <- Set.fromList <$> traverse componentName (Set.toAscList dependencies)+  unique ["component", name, "migrations"] [migration | Migration migration _ _ <- NE.toList migrations]+  pure (MigrationComponent validated deps migrations)++migrationComponentFromList :: Text -> Set Text -> [Migration] -> Either DefinitionError MigrationComponent+migrationComponentFromList name deps migrations = nonEmpty ["component", name, "migrations"] migrations >>= migrationComponent name deps++nonEmpty :: [Text] -> [a] -> Either DefinitionError (NonEmpty a)+nonEmpty path = maybe (Left (DefinitionError EmptyCollection path Nothing)) Right . NE.nonEmpty++unique :: (Ord a) => [Text] -> [a] -> Either DefinitionError ()+unique path names+  | Set.size (Set.fromList names) == length names = Right ()+  | otherwise = Left (DefinitionError DuplicateName path Nothing)
+ src/ClickHouseMigrate/Error.hs view
@@ -0,0 +1,53 @@+-- | Safe, structured errors. Paths identify definitions, never SQL contents.+module ClickHouseMigrate.Error+  ( DefinitionError (..),+    DefinitionErrorCode (..),+    PlanError (..),+    PlanErrorCode (..),+  )+where++import ClickHouseMigrate.Prelude++data DefinitionErrorCode+  = InvalidName+  | EmptyCollection+  | DuplicateName+  | InvalidUtf8+  | NulInSql+  | EmptySql+  | UnterminatedSql+  | MultipleStatements+  | UnsupportedSql+  | ProtectedSetting+  | UnknownChecksumFormat+  deriving stock (Eq, Ord, Show)++-- | Offset, when present, is zero-based Unicode character offset in SQL.+data DefinitionError = DefinitionError+  { code :: !DefinitionErrorCode,+    path :: ![Text],+    characterOffset :: !(Maybe Int)+  }+  deriving stock (Eq, Show)++data PlanErrorCode+  = EmptyPlan+  | DuplicateComponent+  | MissingDependency+  | SelfDependency+  | DependencyCycle+  | EmptySchedule+  | UnknownScheduledMigration+  | DuplicateScheduledMigration+  | MissingScheduledMigration+  | MigrationOrderChanged+  | DependencyNotActivated+  | DependencyAfterDependent+  deriving stock (Eq, Ord, Show)++data PlanError = PlanError+  { code :: !PlanErrorCode,+    components :: ![Text]+  }+  deriving stock (Eq, Show)
+ src/ClickHouseMigrate/Event.hs view
@@ -0,0 +1,26 @@+-- | Observations cannot replace durable ledger audit history. A throwing sink+-- stops the operation. No default sink prints or retains SQL or credentials.+module ClickHouseMigrate.Event+  ( EventPhase (..),+    RunEvent (..),+    EventSink,+    quietEvents,+    DiagnosticPolicy,+    redactedDiagnostics,+    sqlDiagnostics,+    SqlDiagnostic (..),+  )+where++import ClickHouseMigrate.Internal.EventTypes++quietEvents :: EventSink+quietEvents _ = pure ()++redactedDiagnostics :: DiagnosticPolicy+redactedDiagnostics = DiagnosticPolicy Nothing++-- | Explicit opt-in delivery of authored SQL. The caller owns logging/storage.+-- Structured errors always retain safe codes rather than raw server messages.+sqlDiagnostics :: (SqlDiagnostic -> IO ()) -> DiagnosticPolicy+sqlDiagnostics = DiagnosticPolicy . Just
+ src/ClickHouseMigrate/ExclusiveAccess.hs view
@@ -0,0 +1,46 @@+-- | Explicit integration with the application's administrative coordinator.+module ClickHouseMigrate.ExclusiveAccess+  ( ExclusiveAccess,+    externallySerialized,+    AccessDescription,+    exclusiveAccessDescription,+    accessStrategy,+    accessScope,+    rejectAccessAcquisition,+    AccessError (..),+  )+where++import ClickHouseMigrate.Internal.ExclusiveTypes+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (TargetKey)+import Control.Exception (throwIO)+import Data.Text qualified as Text++-- | Acquire excludes ALL administrators for the complete target, and must be+-- exception safe. Release must not admit a successor until this client cannot+-- submit and all predecessor server work, including ledger INSERTs/DDL, is+-- quiescent. Returning True from 'alive' attests the lease is still held; a lost+-- lease cannot fence already submitted SQL. Types cannot verify these external+-- promises. There is deliberately no default/no-op provider.+externallySerialized :: Text -> Text -> Int -> (TargetKey -> IO lease) -> (lease -> IO ()) -> (lease -> IO Bool) -> Either AccessError ExclusiveAccess+externallySerialized strategy scope acquisitionMicros acquire release alive+  | valid strategy && valid scope && acquisitionMicros > 0 && acquisitionMicros <= 300000000 =+      Right (ExclusiveAccess (AccessDescription strategy scope) acquisitionMicros acquire release alive)+  | otherwise = Left InvalidAccessStrategy+  where+    valid value = not (Text.null (Text.strip value)) && Text.length value <= 1024 && not (Text.any (== '\0') value)++accessStrategy :: AccessDescription -> Text+accessStrategy (AccessDescription strategy _) = strategy++accessScope :: AccessDescription -> Text+accessScope (AccessDescription _ scope) = scope++-- | Reject acquisition with a safe typed code from a caller-owned provider.+rejectAccessAcquisition :: AccessError -> IO a+rejectAccessAcquisition = throwIO . AccessAcquisitionException++-- | Read-only description; this does not acquire or expose a live guard.+exclusiveAccessDescription :: ExclusiveAccess -> AccessDescription+exclusiveAccessDescription (ExclusiveAccess description _ _ _ _) = description
+ src/ClickHouseMigrate/Inspection.hs view
@@ -0,0 +1,76 @@+-- | Read-only observations. No inspection path initializes or repairs history.+module ClickHouseMigrate.Inspection+  ( InspectionOptions (Observe, Guarded),+    VerifyOptions,+    verifyOptions,+    status,+    history,+    verify,+  )+where++import ClickHouseMigrate.Internal.Exclusive (checkGuard, withExclusive)+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessResult (..), ExclusiveAccess (..))+import ClickHouseMigrate.Internal.Identity (freshUuid)+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Internal.RuntimeTypes (InspectionOptions (..), VerifyOptions (..))+import ClickHouseMigrate.Internal.ServerPolicy (TargetCheckError (..), withTransportChecked)+import ClickHouseMigrate.Internal.TargetTypes (QueryId (..), TargetConfig (..))+import ClickHouseMigrate.Internal.Transport qualified as Transport+import ClickHouseMigrate.Plan (MigrationPlan)+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R+import Control.Exception (SomeAsyncException, SomeException, fromException, throwIO, try)+import Data.IORef (newIORef, readIORef, writeIORef)++verifyOptions :: Bool -> VerifyOptions+verifyOptions = VerifyOptions++history :: InspectionOptions -> TargetConfig -> IO (Either R.InspectionFailure R.HistoryReport)+history options config@(TargetConfig target _ _ _ _ _ _) = do+  latest <- newIORef Nothing+  observed <- try @SomeException $ do+    probe <- QueryId . ("chm-" <>) <$> freshUuid+    let scoped guard access =+          let check = case guard of Nothing -> pure (Right ()); Just capability -> checkGuard capability >>= pure . either (const (Left (Transport.TransportError Transport.AdministrativeAccessLost Nothing Nothing))) (const (Right ()))+           in withTransportChecked config probe check $ \transport -> Store.storeResult $ do+                observation <- Store.load guard transport config+                let report = Reporting.observationReport target (Just (Transport.transportIdentity transport)) access observation+                writeIORef latest (Just report)+                pure report+        flatten (Transport.ScopeResult result cleanup) =+          ( case result of+              Left failure -> Left (R.TargetFailed failure)+              Right (Left (TargetRejected failure)) -> Left (R.TargetFailed failure)+              Right (Left (ProbeFailure failure)) -> Left (Reporting.outcomeError failure Nothing)+              Right (Right (Left failure)) -> Left (Reporting.storeError failure)+              Right (Right (Right report)) -> Right report,+            map R.TransportCleanup cleanup+          )+    case options of+      Observe -> flatten <$> scoped Nothing R.UnguardedObservation+      Guarded provider@(ExclusiveAccess description _ _ _ _) -> do+        AccessResult result cleanup <- withExclusive provider target (\guard -> scoped (Just guard) (R.GuardedObservation description))+        pure $ case result of+          Left failure -> (Left (R.AccessFailed failure), map R.AccessCleanup cleanup)+          Right value -> let (answer, errors) = flatten value in (answer, errors ++ map R.AccessCleanup cleanup)+  snapshot <- readIORef latest+  case observed of+    Left exception | isJust (fromException exception :: Maybe SomeAsyncException) -> throwIO exception+    Left _ -> pure (Left (R.InspectionFailure R.UnexpectedRuntimeFailure snapshot []))+    Right (Left failure, cleanup) -> pure (Left (R.InspectionFailure failure snapshot cleanup))+    Right (Right _, cleanup@(_ : _)) -> pure (Left (R.InspectionFailure R.CleanupFailed snapshot cleanup))+    Right (Right report, []) -> pure (Right report)++status :: InspectionOptions -> TargetConfig -> MigrationPlan -> IO (Either R.InspectionFailure R.InspectionReport)+status options config plan = fmap (Reporting.inspectHistory plan) <$> history options config++verify :: InspectionOptions -> TargetConfig -> VerifyOptions -> MigrationPlan -> IO (Either R.InspectionFailure R.VerificationReport)+verify options config (VerifyOptions requireCurrent) plan = do+  result <- status options config plan+  pure $+    result >>= \report@(R.InspectionReport snapshot _ _) -> case Reporting.compatibilityError report of+      Just failure -> Left (R.InspectionFailure failure (Just snapshot) [])+      Nothing | requireCurrent && not (Reporting.isCurrent report) -> Left (R.InspectionFailure R.VerificationNotCurrent (Just snapshot) [])+      Nothing -> Right (R.VerificationReport report (Reporting.isCurrent report))
+ src/ClickHouseMigrate/Internal/AdoptionTypes.hs view
@@ -0,0 +1,51 @@+-- | Private representations. Adoption input cannot bypass validated construction.+module ClickHouseMigrate.Internal.AdoptionTypes where++import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (TargetKey)++data AdoptionError+  = InvalidSourceEvidence+  | UnsupportedSourceChecksum+  | InvalidSourceChecksum+  | DuplicateSourceRow+  | DuplicateMapping+  | IncompleteMapping+  | UnknownSourceRow+  | MappingOrderMismatch+  | MappingDefinitionMismatch+  | HistoricalChecksumMismatch+  | HistoricalChecksumRequired+  | AdoptionTargetMismatch+  | AdoptionInitializationRequired+  | AdoptionExpectedStateMismatch+  | AdoptionNativeConflict+  | AdoptionBatchConflict+  | AdoptionResumeRequired+  | AdoptionBatchTooLarge+  deriving stock (Eq, Show)++data SourceChecksum = SourceChecksum !Text !Text deriving stock (Eq, Show)++data LegacyRow = LegacyRow !Text !Text !(Maybe SourceChecksum) deriving stock (Eq, Show)++-- Historical source bytes deliberately have no Show instance and never enter+-- default reports. They are used only to verify an actual predecessor checksum.+data MappingEvidence = Attested ![Text] | Verified !ByteString ![Text]++data AdoptionMapping = AdoptionMapping !Text !Text !Text !Text !MappingEvidence++data AdoptionRequest+  = AdoptionRequest+      !TargetKey+      !L.ExpectedLedgerState+      !(Maybe L.InitializationAuthorization)+      !L.QuiescenceEvidence+      !L.AuditContext+      !Text+      !Text+      ![LegacyRow]+      ![AdoptionMapping]++data AdoptionPlan = AdoptionPlan !AdoptionRequest !L.AdoptionBatch ![(L.Definition, L.Evidence)]
+ src/ClickHouseMigrate/Internal/AdoptionValidation.hs view
@@ -0,0 +1,174 @@+-- | Versioned whole-batch validation. No IO, SQL splitting or schema inference.+module ClickHouseMigrate.Internal.AdoptionValidation where++import ClickHouseMigrate.Definition qualified as Def+import ClickHouseMigrate.Internal.AdoptionTypes+import ClickHouseMigrate.Internal.Checksum qualified as Hash+import ClickHouseMigrate.Internal.LedgerCodec qualified as Codec+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Target (TargetKey, ledgerId)+import Data.ByteString qualified as BS+import Data.ByteString.Builder (Builder, byteString, toLazyByteString, word64BE)+import Data.ByteString.Lazy qualified as Lazy+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding++require :: AdoptionError -> Bool -> Either AdoptionError ()+require failure condition = if condition then Right () else Left failure++validText :: Int -> Text -> Bool+validText bound value = not (Text.null (Text.strip value)) && Text.length value <= bound && not (Text.any (`elem` ['\0', '\n', '\r']) value)++validHash :: Text -> Bool+validHash value = Text.length value == 64 && Text.all (\c -> c >= '0' && c <= '9' || c >= 'a' && c <= 'f') value++references :: [Text] -> Either AdoptionError ()+references refs = do+  require InvalidSourceEvidence (not (null refs) && length refs <= 24)+  either (const (Left InvalidSourceEvidence)) (const (Right ())) (Ledger.auditContext "adoption evidence" "reviewed mapping" refs)++sourceChecksum :: Text -> Text -> Either AdoptionError SourceChecksum+sourceChecksum algorithm digest = do+  require UnsupportedSourceChecksum (algorithm == "sha256")+  require InvalidSourceChecksum (validHash digest)+  pure (SourceChecksum algorithm digest)++legacyRow :: Text -> Text -> Maybe SourceChecksum -> Either AdoptionError LegacyRow+legacyRow identity appliedAt checksum = do+  require InvalidSourceEvidence (validText 512 identity && validText 256 appliedAt)+  pure (LegacyRow identity appliedAt checksum)++mapping :: Text -> Text -> Text -> Text -> MappingEvidence -> Either AdoptionError AdoptionMapping+mapping row component migration checksum evidence = do+  require InvalidSourceEvidence (validText 512 row && validHash checksum)+  either (const (Left MappingDefinitionMismatch)) (const (Right ())) (Def.componentName component)+  either (const (Left MappingDefinitionMismatch)) (const (Right ())) (Def.migrationName migration)+  case evidence of+    Attested refs -> references refs+    Verified bytes refs -> require AdoptionBatchTooLarge (BS.length bytes <= 2097152) >> references refs+  pure (AdoptionMapping row component migration checksum evidence)++adoptionRequest :: TargetKey -> L.ExpectedLedgerState -> Maybe L.InitializationAuthorization -> L.QuiescenceEvidence -> L.AuditContext -> Text -> Text -> NE.NonEmpty LegacyRow -> NE.NonEmpty AdoptionMapping -> Either AdoptionError AdoptionRequest+adoptionRequest target expected authorization quiet audit source reference rows mappings = do+  require InvalidSourceEvidence (validText 512 source && validText 1000 reference && not (null (Ledger.auditReferences audit)))+  require AdoptionBatchTooLarge (NE.length rows <= 1000 && NE.length mappings <= 1000)+  let historicalBytes = sum [BS.length bytes | AdoptionMapping _ _ _ _ (Verified bytes _) <- NE.toList mappings]+      referenceBytes = sum [sum (map (BS.length . Encoding.encodeUtf8) refs) | AdoptionMapping _ _ _ _ evidence <- NE.toList mappings, refs <- [case evidence of Attested values -> values; Verified _ values -> values]]+  require AdoptionBatchTooLarge (historicalBytes <= 67108864 && referenceBytes <= 8388608)+  require AdoptionTargetMismatch (Ledger.quiescenceTarget quiet == target)+  case expected of+    L.ExpectAbsent actual -> require AdoptionTargetMismatch (actual == target) >> require AdoptionInitializationRequired (isJust authorization)+    L.ExpectPosition (L.LedgerPosition expectedLedger _ _) -> case authorization of+      Just (L.InitializationAuthorization _ authorizedLedger _) -> require AdoptionTargetMismatch (expectedLedger == authorizedLedger)+      _ -> pure ()+  case authorization of Just (L.InitializationAuthorization actual _ _) -> require AdoptionTargetMismatch (actual == target); _ -> pure ()+  pure (AdoptionRequest target expected authorization quiet audit source reference (NE.toList rows) (NE.toList mappings))++field :: ByteString -> Builder+field bytes = word64BE (fromIntegral (BS.length bytes)) <> byteString bytes++textField :: Text -> Builder+textField = field . Encoding.encodeUtf8++list :: (a -> Builder) -> [a] -> Builder+list encode values = word64BE (fromIntegral (length values)) <> foldMap encode values++hashEncoding :: ByteString -> Builder -> Text+hashEncoding domain value = Hash.digestHex (Hash.digest (Lazy.toStrict (toLazyByteString (byteString domain <> value))))++snapshotDigest :: Text -> Text -> [LegacyRow] -> Text+snapshotDigest source reference rows = hashEncoding "clickhouse-migrate/adoption-snapshot\0v1\0" (textField source <> textField reference <> list encode rows)+  where+    encode (LegacyRow identity appliedAt checksum) = textField identity <> textField appliedAt <> case checksum of Nothing -> word64BE 0; Just (SourceChecksum algorithm digest) -> word64BE 1 <> textField algorithm <> textField digest++mappingDigest :: Text -> [(Text, L.Definition, L.Evidence)] -> Text+mappingDigest snapshot entries = hashEncoding "clickhouse-migrate/adoption-mapping\0v1\0" (textField snapshot <> list encode entries)+  where+    encode (row, definition, L.Evidence level refs) = textField row <> field (Codec.encodeValue (Codec.definitionValue definition)) <> word64BE (case level of L.OperatorAttestation -> 0; L.PredecessorChecksum -> 1) <> list textField refs++buildBatch :: Plan.MigrationPlan -> AdoptionRequest -> Either AdoptionError (L.AdoptionBatch, [(L.Definition, L.Evidence)])+buildBatch plan (AdoptionRequest _ _ _ _ audit source reference rows mappings) = do+  let rowIds = [identity | LegacyRow identity _ _ <- rows]+      mappingIds = [identity | AdoptionMapping identity _ _ _ _ <- mappings]+      nativeIds = [(component, migration) | AdoptionMapping _ component migration _ _ <- mappings]+      supplied = Codec.definitions plan+  require DuplicateSourceRow (Set.size (Set.fromList rowIds) == length rowIds)+  require DuplicateMapping (Set.size (Set.fromList mappingIds) == length mappingIds && Set.size (Set.fromList nativeIds) == length nativeIds)+  require UnknownSourceRow (all (`elem` rowIds) mappingIds)+  require IncompleteMapping (Set.fromList rowIds == Set.fromList mappingIds && length mappings <= length supplied)+  require MappingOrderMismatch (rowIds == mappingIds)+  identity <- either (const (Left InvalidSourceEvidence)) Right (ledgerId "00000000-0000-4000-8000-000000000000")+  entries <- sequence [entry row mapped definition | (row, mapped, definition) <- zip3 rows mappings supplied]+  let snapshot = snapshotDigest source reference rows+      mapped = mappingDigest snapshot entries+      definitions = [(definition, evidence) | (_, definition, evidence) <- entries]+      batch = L.AdoptionBatch (source <> "\nsnapshot-evidence:" <> reference) snapshot mapped (Codec.definitionsDigest (map fst definitions)) (length definitions)+      uuid = "00000000-0000-4000-8000-000000000000"+      event body context = L.LedgerEvent 1 uuid identity uuid Nothing "2026-10-16T00:00:00.999999999999Z" audit "0.1.0.0" context body+  -- Validate each full serialized entry before initialization. Descriptors may+  -- approach the wire bound even when individual user fields are valid.+  mapM_+    ( \(definition, evidence) -> do+        context <- either (const (Left MappingDefinitionMismatch)) Right (Codec.contextFor definition Nothing)+        let encoded = event (L.AdoptEntry definition evidence) (Just context)+        either (const (Left InvalidSourceEvidence)) pure (Codec.validateEvent encoded)+        require AdoptionBatchTooLarge (BS.length (Codec.encodeEvent encoded) <= 61440)+    )+    definitions+  mapM_+    ( \body -> do+        let encoded = event body Nothing+        either (const (Left InvalidSourceEvidence)) pure (Codec.validateEvent encoded)+        require AdoptionBatchTooLarge (BS.length (Codec.encodeEvent encoded) <= 61440)+    )+    [L.StartAdoption batch, L.CompleteAdoption mapped]+  pure (batch, definitions)+  where+    entry (LegacyRow row _ checksum) (AdoptionMapping mapped component migration digest evidence) definition = do+      require MappingOrderMismatch (row == mapped)+      require MappingDefinitionMismatch (component == L.component definition && migration == L.migration definition && digest == L.checksum definition)+      let sourceRefs = ["source:" <> source, "snapshot-evidence:" <> reference, "source-row:" <> row]+      accepted <- case evidence of+        Attested refs -> pure (L.Evidence L.OperatorAttestation (sourceRefs ++ refs))+        Verified bytes refs -> case checksum of+          Nothing -> Left HistoricalChecksumRequired+          Just (SourceChecksum algorithm historical) -> do+            require HistoricalChecksumMismatch (algorithm == "sha256" && Hash.digestHex (Hash.digest bytes) == historical)+            pure (L.Evidence L.PredecessorChecksum (sourceRefs ++ ["historical-algorithm:" <> algorithm, "historical-checksum:" <> historical] ++ refs))+      pure (row, definition, accepted)++-- An identical committed batch is inspection-only, including after later native+-- execution. Every mutating path must match its fresh expected observation.+checkHistory :: AdoptionRequest -> L.AdoptionBatch -> [(L.Definition, L.Evidence)] -> R.HistoryReport -> Either AdoptionError Bool+checkHistory (AdoptionRequest target expected authorization _ _ _ _ _ _) batch entries (R.HistoryReport observed _ presence position _ events migrations unresolved) = do+  require AdoptionTargetMismatch (target == observed)+  case (expected, position) of+    (L.ExpectPosition (L.LedgerPosition wanted _ _), Just (L.LedgerPosition actual _ _)) -> require AdoptionTargetMismatch (wanted == actual)+    _ -> pure ()+  case (authorization, position) of+    (Just (L.InitializationAuthorization _ wanted _), Just (L.LedgerPosition actual _ _)) -> require AdoptionTargetMismatch (wanted == actual)+    _ -> pure ()+  let batches = [fromSummary summary | R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.AdoptionBeginning summary) <- events]+      committed = [completion | R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.AdoptionCompletion completion) <- events]+      actual = [(definition, level, refs) | R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.AdoptedDefinition definition (R.EvidenceSummary level refs)) <- events]+      wanted = [(Reporting.summaryDefinition definition, level, refs) | (definition, L.Evidence level refs) <- entries]+      L.AdoptionBatch _ _ digest _ _ = batch+  if batches == [batch] && committed == [digest]+    then require AdoptionBatchConflict (actual == wanted) >> Right True+    else do+      require AdoptionExpectedStateMismatch (case expected of L.ExpectAbsent _ -> presence `elem` [R.LedgerMissing, R.LedgerUninitialized] && isNothing position; L.ExpectPosition value -> position == Just value)+      require AdoptionNativeConflict (null [() | R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.MigrationDefinition _) <- events])+      case batches of+        [] -> require AdoptionNativeConflict (null migrations && not unresolved)+        [existing] -> require AdoptionBatchConflict (existing == batch && null committed && actual == take (length actual) wanted)+        _ -> Left AdoptionBatchConflict+      pure False+  where+    fromSummary (R.AdoptionSummary source snapshot mapped definitions count) = L.AdoptionBatch source snapshot mapped definitions count
+ src/ClickHouseMigrate/Internal/Checksum.hs view
@@ -0,0 +1,53 @@+-- | Versioned byte encoder. Keep in agreement with docs/user/contracts/checksum-v1.md.+module ClickHouseMigrate.Internal.Checksum+  ( completionPolicyV1,+    encodeStepV1,+    encodeMigrationV1,+    digest,+    digestHex,+  )+where++import ClickHouseMigrate.Prelude+import Crypto.Hash.SHA256 qualified as SHA256+import Data.ByteString qualified as BS+import Data.ByteString.Builder (Builder, byteString, toLazyByteString, word64BE)+import Data.ByteString.Lazy qualified as Lazy+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text+import Numeric (showHex)++completionPolicyV1 :: ByteString+completionPolicyV1 = "clickhouse-migrate/local-sync-v1;async_insert=0;mutations_sync=1;alter_sync=1"++field :: ByteString -> Builder+field bs = word64BE (fromIntegral (BS.length bs)) <> byteString bs++encodeStepV1 :: Text -> Text -> Text -> ByteString -> ByteString+encodeStepV1 component migration step sql =+  Lazy.toStrict $+    toLazyByteString $+      byteString "clickhouse-migrate/step\0v1\0"+        <> field completionPolicyV1+        <> foldMap (field . Text.encodeUtf8) [component, migration, step]+        <> field sql++encodeMigrationV1 :: Text -> Text -> [ByteString] -> ByteString+encodeMigrationV1 component migration steps =+  Lazy.toStrict $+    toLazyByteString $+      byteString "clickhouse-migrate/migration\0v1\0"+        <> field completionPolicyV1+        <> foldMap (field . Text.encodeUtf8) [component, migration]+        <> word64BE (fromIntegral (length steps))+        <> foldMap field steps++digest :: ByteString -> ByteString+digest = SHA256.hash++digestHex :: ByteString -> Text+digestHex = Text.pack . concatMap hex . BS.unpack+  where+    hex byte = case showHex byte "" of+      [c] -> ['0', c]+      cs -> cs
+ src/ClickHouseMigrate/Internal/EventTypes.hs view
@@ -0,0 +1,48 @@+-- | Observational callback configuration. No SQL execution capability is passed.+module ClickHouseMigrate.Internal.EventTypes where++import ClickHouseMigrate.Ledger (LedgerPosition)+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report (StepReference)+import ClickHouseMigrate.Target (QueryId)++data EventPhase+  = RunBeginning+  | RunCompleted+  | BeforeMigrationStart+  | MigrationStarted+  | BeforeStepStart+  | StepStarted+  | BeforeSql+  | SqlObserved+  | BeforeStepSuccess+  | StepSucceeded+  | BeforeMigrationComplete+  | MigrationCompleted+  | RecoveryAuthorized+  | StepAccepted+  | BeforeAdoptionStart+  | AdoptionStarted+  | BeforeAdoptionEntry+  | AdoptionEntryRecorded+  | BeforeAdoptionComplete+  | AdoptionCompleted+  deriving stock (Eq, Ord, Show)++data RunEvent = RunEvent+  { phase :: !EventPhase,+    runId :: !Text,+    component :: !(Maybe Text),+    migration :: !(Maybe Text),+    step :: !(Maybe Text),+    queryId :: !(Maybe QueryId),+    position :: !(Maybe LedgerPosition)+  }+  deriving stock (Eq, Show)++type EventSink = RunEvent -> IO ()++-- Intentionally no Show instance: SQL enters only an explicitly configured sink.+data SqlDiagnostic = SqlDiagnostic !StepReference !QueryId !ByteString++data DiagnosticPolicy = DiagnosticPolicy !(Maybe (SqlDiagnostic -> IO ()))
+ src/ClickHouseMigrate/Internal/Exclusive.hs view
@@ -0,0 +1,51 @@+-- | Scope ownership encloses transport cleanup. Cancellation is always rethrown.+module ClickHouseMigrate.Internal.Exclusive (withExclusive, checkGuard, guardDescription) where++import ClickHouseMigrate.Internal.ExclusiveTypes+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (TargetKey)+import Control.Exception (SomeAsyncException, SomeException, fromException, mask, throwIO, try)+import Data.IORef (newIORef, readIORef, writeIORef)+import System.Timeout (timeout)++withExclusive :: ExclusiveAccess -> TargetKey -> (Guard -> IO a) -> IO (AccessResult a)+withExclusive (ExclusiveAccess description micros acquire release alive) target action = mask $ \restore -> do+  acquired <- try @SomeException (timeout micros (acquire target))+  case acquired of+    Left exception | asynchronous exception -> throwIO exception+    Left exception -> pure (AccessResult (Left (case fromException exception of Just (AccessAcquisitionException failure) -> failure; _ -> AcquisitionFailed)) [])+    Right Nothing -> pure (AccessResult (Left AcquisitionTimedOut) [])+    Right (Just lease) -> do+      active <- newIORef True+      lost <- newIORef False+      let check = do+            live <- readIORef active+            previouslyLost <- readIORef lost+            if not live+              then pure (Left AccessScopeClosed)+              else+                if previouslyLost+                  then pure (Left AccessLost)+                  else do+                    observation <- try @SomeException (timeout 1000000 (alive lease))+                    case observation of+                      Left exception | asynchronous exception -> throwIO exception+                      Right (Just True) -> pure (Right ())+                      _ -> writeIORef lost True >> pure (Left AccessLost)+      result <- try @SomeException (restore (action (Guard description check)))+      writeIORef active False+      released <- try @SomeException (timeout 1000000 (release lease))+      let cleanup = case released of Right (Just ()) -> []; Right Nothing -> [AccessReleaseTimedOut]; Left _ -> [AccessReleaseFailed]+      case result of+        Left exception | asynchronous exception -> throwIO exception+        _ -> case released of+          Left exception | asynchronous exception -> throwIO exception+          _ -> pure (AccessResult (either (const (Left AccessActionFailed)) Right result) cleanup)+  where+    asynchronous exception = isJust (fromException exception :: Maybe SomeAsyncException)++checkGuard :: Guard -> IO (Either AccessError ())+checkGuard (Guard _ check) = check++guardDescription :: Guard -> AccessDescription+guardDescription (Guard description _) = description
+ src/ClickHouseMigrate/Internal/ExclusiveTypes.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE ExistentialQuantification #-}++-- | Private acquisition and live-scope capability.+module ClickHouseMigrate.Internal.ExclusiveTypes where++import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (TargetKey)+import Control.Exception (Exception)++data AccessError = InvalidAccessStrategy | AcquisitionTimedOut | AcquisitionFailed | AccessLost | AccessScopeClosed | AccessActionFailed | PredecessorQuiescenceRequired | AccessReleaseFailed | AccessReleaseTimedOut+  deriving stock (Eq, Ord, Show)++data AccessDescription = AccessDescription !Text !Text deriving stock (Eq, Show)++data ExclusiveAccess+  = forall lease. ExclusiveAccess+      !AccessDescription+      !Int+      !(TargetKey -> IO lease)+      !(lease -> IO ())+      !(lease -> IO Bool)++data Guard = Guard !AccessDescription !(IO (Either AccessError ()))++data AccessResult a = AccessResult !(Either AccessError a) ![AccessError]+  deriving stock (Eq, Show)++data AccessAcquisitionException = AccessAcquisitionException !AccessError deriving stock (Show)++instance Exception AccessAcquisitionException
+ src/ClickHouseMigrate/Internal/Execution.hs view
@@ -0,0 +1,280 @@+-- | One scoped administrative operation. No live capability is public.+module ClickHouseMigrate.Internal.Execution where++import ClickHouseMigrate.Event qualified as Event+import ClickHouseMigrate.Internal.EventTypes (DiagnosticPolicy (..), SqlDiagnostic (..))+import ClickHouseMigrate.Internal.Exclusive (checkGuard, guardDescription, withExclusive)+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessDescription (..), AccessResult (..), ExclusiveAccess (..), Guard)+import ClickHouseMigrate.Internal.Identity (diagnosticTime, freshUuid)+import ClickHouseMigrate.Internal.LedgerCodec (contextFor, definitions)+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Preflight (preflight, validateRecoveryEvent)+import ClickHouseMigrate.Internal.Projection (appendProjection)+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Internal.RuntimeTypes+import ClickHouseMigrate.Internal.ServerPolicy (TargetCheckError (..), withTransportChecked)+import ClickHouseMigrate.Internal.TargetTypes qualified as T+import ClickHouseMigrate.Internal.Transport qualified as Transport+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Target (queryIdText)+import Control.Exception (Exception, SomeAsyncException, SomeException, fromException, mask, throwIO, try)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.List.NonEmpty qualified as NE+import Data.Text qualified as Text++-- Configuration, audit, plan, scope, transport, store, run ID, latest ack and+-- secondary observations are owned by one synchronous command invocation.+data Runtime = Runtime !MutationContext !L.AuditContext !Plan.MigrationPlan !Guard !Transport.Transport !Store.LedgerStore !Text !(IORef R.RunReport) !(IORef [R.SecondaryError])++data RunAbort = RunAbort !R.RunError deriving stock (Show)++instance Exception RunAbort++abort :: R.RunError -> IO a+abort = throwIO . RunAbort++callStore :: IO a -> IO a+callStore action = Store.storeResult action >>= either (abort . Reporting.storeError) pure++isAsync :: SomeException -> Bool+isAsync exception = isJust (fromException exception :: Maybe SomeAsyncException)++tryAction :: IO a -> IO (Either R.RunError a)+tryAction action =+  try @SomeException action >>= \case+    Left exception | isAsync exception -> throwIO exception+    Left exception -> pure (Left (case fromException exception of Just (RunAbort failure) -> failure; _ -> R.UnexpectedRuntimeFailure))+    Right value -> pure (Right value)++withRuntime :: MutationContext -> L.AuditContext -> Maybe L.InitializationAuthorization -> Maybe L.QuiescenceEvidence -> Plan.MigrationPlan -> (Runtime -> IO ()) -> IO (Either R.RunFailure R.RunReport)+withRuntime = withRuntimeCheck (Right ())++withRuntimeCheck :: Either R.RunError () -> MutationContext -> L.AuditContext -> Maybe L.InitializationAuthorization -> Maybe L.QuiescenceEvidence -> Plan.MigrationPlan -> (Runtime -> IO ()) -> IO (Either R.RunFailure R.RunReport)+withRuntimeCheck prerequisite = withRuntimeObservation prerequisite Nothing++-- Adoption validates its fresh read before openStore may initialize anything.+-- Ordinary execution retains its existing single-load path.+withRuntimeObservation :: Either R.RunError () -> Maybe (R.HistoryReport -> Either R.RunError Bool) -> MutationContext -> L.AuditContext -> Maybe L.InitializationAuthorization -> Maybe L.QuiescenceEvidence -> Plan.MigrationPlan -> (Runtime -> IO ()) -> IO (Either R.RunFailure R.RunReport)+withRuntimeObservation prerequisite validateObservation context@(MutationContext config@(T.TargetConfig target _ _ _ _ _ _) provider@(ExclusiveAccess description _ _ _ _) _ _) audit initialization quiet plan action = do+  seed <- try @SomeException freshUuid+  let identity = either (const "00000000-0000-4000-8000-000000000000") id seed+  latest <- newIORef (Reporting.initialRun identity target description plan)+  secondary <- newIORef []+  result <- case seed of+    Left exception | isAsync exception -> throwIO exception+    Left _ -> pure (Left R.UnexpectedRuntimeFailure, [])+    Right _ -> do+      checked <- tryAction $ do+        either abort pure prerequisite+        either abort pure (preflight config audit plan)+        case initialization of+          Just (L.InitializationAuthorization authorized _ authAudit) -> do+            unless (authorized == target) (abort (R.LedgerFailed L.InitializationNotAuthorized))+            either abort pure (preflight config authAudit plan)+          Nothing -> pure ()+        emitAt context identity latest Event.RunBeginning Nothing Nothing Nothing+      case checked of+        Left failure -> pure (Left failure, [])+        Right () -> do+          scoped <- tryAction $ withExclusive provider target $ \guard -> do+            modifyIORef' latest (\report -> report {R.acquired = True})+            probe <- T.QueryId . ("chm-" <>) <$> freshUuid+            let check = checkGuard guard >>= pure . either (const (Left (Transport.TransportError Transport.AdministrativeAccessLost Nothing Nothing))) (const (Right ()))+            withTransportChecked config probe check $ \transport -> tryAction $ do+              let server = Just (Transport.transportIdentity transport)+              modifyIORef'+                latest+                ( \report -> case Reporting.runHistory report of+                    R.HistoryReport key _ presence position access events migrations unresolved -> Reporting.updateRun plan (R.HistoryReport key server presence position access events migrations unresolved) report+                )+              let AccessDescription strategy scope = guardDescription guard+                  -- The caller's coordinator contract is the authority for a new+                  -- ordinary job. Repair supplies a separate explicit operator+                  -- attestation; no elapsed-time/process-list inference is made.+                  established = fromMaybe (L.QuiescenceEvidence target (L.AuditContext (Ledger.auditActor audit) "Coordinator granted a job after all predecessor requests became quiescent" ["strategy:" <> strategy, "scope:" <> scope])) quiet+              observationOnly <- case validateObservation of+                Nothing -> pure False+                Just validate -> do+                  observation <- callStore (Store.load (Just guard) transport config)+                  let snapshot = Reporting.observationReport target server (R.GuardedObservation (guardDescription guard)) observation+                  modifyIORef' latest (Reporting.updateRun plan snapshot)+                  either abort pure (validate snapshot)+              if observationOnly+                then do+                  emitAt context identity latest Event.RunCompleted Nothing Nothing Nothing+                  callStore (Store.checkAccess guard)+                else do+                  opened <- Store.storeResult (Store.openStore guard transport config initialization established identity)+                  store <- case opened of+                    Left (Store.LedgerProblem L.InitializationNotAuthorized) | T.TargetConfig _ _ _ _ (T.ExistingLedger _) _ _ <- config, isNothing initialization -> abort R.LedgerRequired+                    Left failure -> abort (Reporting.storeError failure)+                    Right value -> pure value+                  let runtime = Runtime context audit plan guard transport store identity latest secondary+                  refresh runtime+                  case (config, initialization) of+                    (T.TargetConfig _ _ _ _ T.FirstUse _ _, Just (L.InitializationAuthorization _ ledger _)) -> modifyIORef' latest (\report -> report {R.initialized = Just ledger})+                    _ -> pure ()+                  report <- readIORef latest+                  let R.InspectionReport _ _ issues = Reporting.inspectHistory plan (Reporting.runHistory report)+                      changes = [issue | issue@R.DefinitionChanged {} <- issues]+                  unless (null changes) (abort (R.IncompatibleHistory changes))+                  action runtime+                  emit runtime Event.RunCompleted Nothing Nothing Nothing+                  callStore (Store.checkAccess guard)+          pure $ case scoped of+            Left failure -> (Left failure, [])+            Right (AccessResult acquired accessCleanup) -> case acquired of+              Left failure -> (Left (R.AccessFailed failure), map R.AccessCleanup accessCleanup)+              Right (Transport.ScopeResult network transportCleanup) ->+                let outcome = case network of+                      Left failure -> Left (R.TargetFailed failure)+                      Right (Left (TargetRejected failure)) -> Left (R.TargetFailed failure)+                      Right (Left (ProbeFailure failure)) -> Left (Reporting.outcomeError failure Nothing)+                      Right (Right value) -> value+                 in (outcome, map R.TransportCleanup transportCleanup ++ map R.AccessCleanup accessCleanup)+  snapshot <- readIORef latest+  extra <- readIORef secondary+  pure $ case result of+    (Left failure, cleanup) -> Left (R.RunFailure failure snapshot (extra ++ cleanup))+    (Right (), []) | null extra -> Right snapshot+    (Right (), cleanup) -> Left (R.RunFailure R.CleanupFailed snapshot (extra ++ cleanup))++runtimeStore :: Runtime -> Store.LedgerStore+runtimeStore (Runtime _ _ _ _ _ store _ _ _) = store++runtimeTransport :: Runtime -> Transport.Transport+runtimeTransport (Runtime _ _ _ _ transport _ _ _ _) = transport++runtimeGuard :: Runtime -> Guard+runtimeGuard (Runtime _ _ _ guard _ _ _ _ _) = guard++runtimeRunId :: Runtime -> Text+runtimeRunId (Runtime _ _ _ _ _ _ identity _ _) = identity++runtimeSnapshot :: Runtime -> IO R.RunReport+runtimeSnapshot (Runtime _ _ _ _ _ _ _ latest _) = readIORef latest++runtimeProjection :: Runtime -> IO L.Projection+runtimeProjection = callStore . Store.storeProjection . runtimeStore++runtimeAudit :: Runtime -> L.AuditContext+runtimeAudit (Runtime _ audit _ _ _ _ _ _ _) = audit++refresh :: Runtime -> IO ()+refresh runtime@(Runtime _ _ plan guard transport _ _ latest _) = do+  projection <- runtimeProjection runtime+  let snapshot = Reporting.fromProjection (Just (Transport.transportIdentity transport)) (R.GuardedObservation (guardDescription guard)) projection+  modifyIORef' latest (Reporting.updateRun plan snapshot)++emitAt :: MutationContext -> Text -> IORef R.RunReport -> Event.EventPhase -> Maybe L.Definition -> Maybe Int -> Maybe T.QueryId -> IO ()+emitAt (MutationContext _ _ sink _) identity latest phase definition index query = do+  report <- readIORef latest+  let component = fmap L.component definition+      migration = fmap L.migration definition+      step = do def <- definition; i <- index; L.StepDefinition name _ <- Reporting.at i (L.steps def); pure name+      event = Event.RunEvent phase identity component migration step query (Reporting.runPosition report)+  result <- try @SomeException (sink event)+  case result of+    Left exception | isAsync exception -> throwIO exception+    Left _ -> abort (R.CallbackFailed (R.EventCallback (Text.pack (show phase))))+    Right () -> pure ()++emit :: Runtime -> Event.EventPhase -> Maybe L.Definition -> Maybe Int -> Maybe T.QueryId -> IO ()+emit (Runtime context _ _ _ _ _ identity latest _) = emitAt context identity latest++-- Frozen deterministic metadata is reused after recovery preparation.+data PreparedEvent = PreparedEvent !L.LedgerEvent++prepareEvent :: Runtime -> L.Projection -> Maybe Text -> L.EventBody -> Maybe L.EventContext -> IO PreparedEvent+prepareEvent runtime projection attempt body context = do+  let L.LedgerPosition ledger number _ = L.position projection+  when (number == maxBound) (abort (R.LedgerFailed L.LedgerLimitExceeded))+  identity <- freshUuid+  time <- diagnosticTime+  pure (PreparedEvent (L.LedgerEvent (number + 1) identity ledger (runtimeRunId runtime) attempt time (runtimeAudit runtime) "0.1.0.0" context body))++preparedProjection :: R.RecoveryEventKind -> L.Projection -> PreparedEvent -> IO L.Projection+preparedProjection kind projection (PreparedEvent event) = do+  either abort pure (validateRecoveryEvent kind event)+  either (abort . R.LedgerFailed) pure (appendProjection projection event)++appendPreparedEvent :: Runtime -> PreparedEvent -> IO ()+appendPreparedEvent runtime (PreparedEvent event) = mask $ \restore -> do+  _ <- restore (callStore (Store.append (runtimeStore runtime) event))+  refresh runtime++appendEvent :: Runtime -> Maybe Text -> L.EventBody -> Maybe L.EventContext -> IO ()+appendEvent runtime attempt body context = do+  projection <- runtimeProjection runtime+  prepareEvent runtime projection attempt body context >>= appendPreparedEvent runtime++contextAt :: L.Definition -> Maybe Int -> IO (Maybe L.EventContext)+contextAt definition index = Just <$> either (const (abort R.UnexpectedRuntimeFailure)) pure (contextFor definition index)++recordObservation :: Runtime -> R.SqlObservation -> IO ()+recordObservation (Runtime _ _ _ _ _ _ _ latest _) value = modifyIORef' latest (Reporting.recordSql value)++secondaryAction :: Runtime -> (R.RunError -> R.SecondaryError) -> IO () -> IO ()+secondaryAction (Runtime _ _ _ _ _ _ _ _ errors) classify action =+  tryAction action >>= \case+    Left failure -> modifyIORef' errors (++ [classify failure])+    Right () -> pure ()++executeStep :: Runtime -> L.Definition -> Text -> Int -> Plan.StepView -> IO ()+executeStep runtime@(Runtime (MutationContext _ _ _ (DiagnosticPolicy diagnostic)) _ _ _ transport _ _ _ _) definition attempt index step = do+  reference <- maybe (abort R.UnexpectedRuntimeFailure) pure (Reporting.stepReference definition index)+  query <- Store.newQueryId+  context <- contextAt definition (Just index)+  emit runtime Event.BeforeStepStart (Just definition) (Just index) (Just query)+  appendEvent runtime (Just attempt) (L.StartStep index (queryIdText query)) context+  emit runtime Event.StepStarted (Just definition) (Just index) (Just query)+  emit runtime Event.BeforeSql (Just definition) (Just index) (Just query)+  case diagnostic of+    Nothing -> pure ()+    Just sink -> do+      observed <- try @SomeException (sink (SqlDiagnostic reference query (Plan.stepSql step)))+      case observed of Left exception | isAsync exception -> throwIO exception; Left _ -> abort (R.CallbackFailed R.SqlDiagnosticCallback); Right () -> pure ()+  outcome <- Transport.request transport query Transport.Command (Plan.stepSql step)+  case outcome of+    Transport.AcknowledgedSuccess Transport.CommandAccepted -> do+      recordObservation runtime (R.SqlObservation reference query R.QueryAcknowledged)+      emit runtime Event.SqlObserved (Just definition) (Just index) (Just query)+      emit runtime Event.BeforeStepSuccess (Just definition) (Just index) (Just query)+      appendEvent runtime (Just attempt) (L.SucceedStep index (queryIdText query)) context+      emit runtime Event.StepSucceeded (Just definition) (Just index) (Just query)+    _ -> do+      let (evidence, failure) = case outcome of+            Transport.NotSubmitted value -> (Ledger.PositiveNonSubmission, value)+            Transport.ObservedServerFailure value -> (Ledger.ObservedFailure, value)+            Transport.SubmissionUncertain value -> (Ledger.UncertainSubmission, value)+            _ -> (Ledger.UncertainSubmission, Transport.TransportError Transport.UnexpectedResponse Nothing Nothing)+          primary = R.SqlFailed reference query evidence failure+      recordObservation runtime (R.SqlObservation reference query (R.QueryRejected evidence failure))+      when (evidence == Ledger.ObservedFailure) $+        secondaryAction runtime R.RecordingError (appendEvent runtime (Just attempt) (L.FailStep index (queryIdText query) (L.FailureEvidence (Text.pack (show (Transport.code failure))) (Transport.httpStatus failure) (Transport.serverCode failure))) context)+      secondaryAction runtime (\case R.CallbackFailed phase -> R.ObserverError phase; other -> R.RecordingError other) (emit runtime Event.SqlObserved (Just definition) (Just index) (Just query))+      abort primary++executeMigration :: Runtime -> L.Definition -> Plan.MigrationView -> IO ()+executeMigration runtime definition migration = do+  attempt <- freshUuid+  context <- contextAt definition Nothing+  emit runtime Event.BeforeMigrationStart (Just definition) Nothing Nothing+  appendEvent runtime (Just attempt) (L.StartMigration definition) context+  emit runtime Event.MigrationStarted (Just definition) Nothing Nothing+  mapM_ (\(index, step) -> executeStep runtime definition attempt index step) (zip [0 ..] (NE.toList (Plan.migrationSteps migration)))+  finishMigration runtime definition attempt++finishMigration :: Runtime -> L.Definition -> Text -> IO ()+finishMigration runtime definition attempt = do+  emit runtime Event.BeforeMigrationComplete (Just definition) Nothing Nothing+  context <- contextAt definition Nothing+  appendEvent runtime (Just attempt) L.CompleteMigration context+  emit runtime Event.MigrationCompleted (Just definition) Nothing Nothing++orderedMigrations :: Plan.MigrationPlan -> [(L.Definition, Plan.MigrationView)]+orderedMigrations plan = zip (definitions plan) (NE.toList (Plan.inspectSchedule plan))
+ src/ClickHouseMigrate/Internal/Http.hs view
@@ -0,0 +1,281 @@+-- | Dedicated direct HTTP resources, bounded full responses, and no replay.+module ClickHouseMigrate.Internal.Http (Http, withHttp, withHttpAdmission, withHttpAdmissionChecked, withHttpFinalizer, probeConnection, submit, readJsonRowStream) where++import ClickHouseMigrate.Internal.TargetTypes+import ClickHouseMigrate.Internal.Transport hiding (request)+import ClickHouseMigrate.Prelude+import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)+import Control.Concurrent.MVar (MVar, modifyMVar_, newMVar, withMVar)+import Control.Exception (AsyncException (ThreadKilled), Exception, SomeAsyncException, SomeException, bracket, finally, fromException, mask, onException, throwIO, try)+import Data.Aeson (Object, eitherDecodeStrict')+import Data.ByteString qualified as BS+import Data.ByteString.Char8 qualified as Char8+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Client.Internal qualified as Internal+import Network.HTTP.Client.TLS (tlsManagerSettings)+import Network.HTTP.Types.Status (statusCode)+import System.Timeout (timeout)+import Text.Read (readMaybe)++data Http = Http !TargetConfig !HTTP.Manager !(IORef Bool) !(MVar ()) !(IORef Bool) !(IORef [ThreadId])++data AdmissionFailure = AdmissionFailure !TransportError deriving stock (Show)++instance Exception AdmissionFailure++safeError :: TransportErrorCode -> TransportError+safeError c = TransportError c Nothing Nothing++-- http-client's closeManager is a no-op. Track/close every owned socket instead.+-- Async cancellation is rethrown after bounded cleanup; never turn it into success.+withHttp :: TargetConfig -> (Http -> IO a) -> IO (ScopeResult a)+withHttp config = withHttpFinalizer config (pure ())++-- Private qualification hook runs through the real cleanup path after sockets.+withHttpFinalizer :: TargetConfig -> IO () -> (Http -> IO a) -> IO (ScopeResult a)+withHttpFinalizer config finalizer = withHttpOwned config (const (pure (Right ()))) (pure (Right ())) finalizer++withHttpAdmission :: TargetConfig -> (Internal.Connection -> IO (Either TransportError ())) -> (Http -> IO a) -> IO (ScopeResult a)+withHttpAdmission config admit = withHttpAdmissionChecked config admit (pure (Right ()))++withHttpAdmissionChecked :: TargetConfig -> (Internal.Connection -> IO (Either TransportError ())) -> IO (Either TransportError ()) -> (Http -> IO a) -> IO (ScopeResult a)+withHttpAdmissionChecked config admit check = withHttpOwned config admit check (pure ())++withHttpOwned :: TargetConfig -> (Internal.Connection -> IO (Either TransportError ())) -> IO (Either TransportError ()) -> IO () -> (Http -> IO a) -> IO (ScopeResult a)+withHttpOwned config admit check finalizer action = mask $ \restore -> do+  submitted <- newIORef False+  active <- newIORef True+  connections <- newIORef [(0 :: Int, finalizer)]+  nextConnection <- newIORef (1 :: Int)+  inflight <- newIORef []+  gate <- newMVar ()+  let track factory = do+        open <- factory+        pure $ \address host port -> mask $ \restoreOpen -> do+          conn <- restoreOpen (open address host port)+          closeGate <- newMVar False+          connectionId <- atomicModifyIORef' nextConnection (\old -> (old + 1, old))+          let close = modifyMVar_ closeGate $ \closed ->+                if closed+                  then pure True+                  else do+                    Internal.connectionClose conn `onException` writeIORef active False+                    atomicModifyIORef' connections (\old -> (filter ((/= connectionId) . fst) old, ()))+                    pure True+              wrapped =+                conn+                  { Internal.connectionWrite = \bytes -> do+                      live <- readIORef active+                      unless live (fail "HTTP scope closed")+                      check >>= either (throwIO . AdmissionFailure) pure+                      writeIORef submitted True+                      Internal.connectionWrite conn bytes,+                    Internal.connectionClose = close+                  }+          atomicModifyIORef' connections (\old -> ((connectionId, close) : old, ()))+          live <- readIORef active+          unless live (close >> fail "HTTP scope closed")+          admitted <- restoreOpen (admit (conn {Internal.connectionWrite = \bytes -> check >>= either (throwIO . AdmissionFailure) (const (Internal.connectionWrite conn bytes))})) `onException` close+          case admitted of+            Left failure -> close >> throwIO (AdmissionFailure failure)+            Right () -> pure ()+          stillLive <- readIORef active+          unless stillLive (close >> fail "HTTP scope closed")+          pure wrapped+      settings =+        HTTP.managerSetProxy+          HTTP.noProxy+          tlsManagerSettings+            { HTTP.managerConnCount = 1,+              HTTP.managerIdleConnectionCount = 1,+              HTTP.managerRetryableException = const False,+              HTTP.managerRawConnection = track (HTTP.managerRawConnection tlsManagerSettings),+              HTTP.managerTlsConnection = track (HTTP.managerTlsConnection tlsManagerSettings)+            }+  manager <- HTTP.newManager settings+  result <- try @SomeException (restore (action (Http config manager submitted gate active inflight)))+  writeIORef active False+  owner <- myThreadId+  running <- readIORef inflight+  cancelled <- try @SomeException $ timeout 1000000 $ do+    mapM_ (\thread -> unless (thread == owner) (throwTo thread ThreadKilled)) running+    let wait = do+          remaining <- readIORef inflight+          unless (null remaining) (threadDelay 1000 >> wait)+    wait+  closers <- map snd <$> readIORef connections+  closed <- mapM (\close -> try @SomeException (timeout 1000000 close)) closers+  let observations = cancelled : closed+      cleanup =+        concatMap+          ( \case+              Right (Just ()) -> []+              Right Nothing -> [safeError CleanupTimedOut]+              Left _ -> [safeError CleanupFailed]+          )+          observations+      cancelledDuringCleanup = [exception | Left exception <- observations, isJust (fromException exception :: Maybe SomeAsyncException)]+  case result of+    Left exception | isJust (fromException exception :: Maybe SomeAsyncException) -> throwIO exception+    _ | exception : _ <- cancelledDuringCleanup -> throwIO exception+    Left _ -> pure (ScopeResult (Left (safeError ClientScopeFailed)) cleanup)+    Right value -> pure (ScopeResult (Right value) cleanup)++submit :: Http -> QueryId -> RequestKind -> ByteString -> IO TransportOutcome+submit (Http (TargetConfig _ (DirectEndpoint endpoint) _ _ _ (Credentials user password) (HttpLimits micros maxBytes)) manager submitted gate active inflight) (QueryId identity) kind sql = registered inflight $ withMVar gate $ \() -> do+  writeIORef submitted False+  live <- readIORef active+  if not live || BS.null sql || BS.elem 0 sql+    then pure (NotSubmitted (safeError InvalidRequest))+    else do+      result <- try @SomeException $ timeout micros $ do+        request <- makeRequest endpoint user password micros identity kind sql+        HTTP.withResponse request manager $ \response -> do+          let status = statusCode (HTTP.responseStatus response)+              exceptionHeader = lookup "X-ClickHouse-Exception-Code" (HTTP.responseHeaders response)+          if kind == JsonRowStream && status == 200 && isNothing exceptionHeader+            then readJsonRowStream maxBytes (HTTP.brRead (HTTP.responseBody response))+            else do+              bytes <- readBounded maxBytes (HTTP.responseBody response)+              pure $ case bytes of+                Left () -> SubmissionUncertain (safeError ResponseTooLarge)+                Right body -> classify kind status exceptionHeader body+      didSubmit <- readIORef submitted+      case result of+        Left exception | isJust (fromException exception :: Maybe SomeAsyncException) -> throwIO exception+        Left exception | Just (AdmissionFailure failure) <- fromException exception -> pure (if didSubmit then SubmissionUncertain failure else NotSubmitted failure)+        Left _ -> pure $ if didSubmit then SubmissionUncertain (safeError ResponseInterrupted) else NotSubmitted (safeError ConnectionFailed)+        Right Nothing -> pure $ if didSubmit then SubmissionUncertain (safeError DeadlineExpired) else NotSubmitted (safeError DeadlineExpired)+        Right (Just outcome) -> pure outcome++readBounded :: Int -> HTTP.BodyReader -> IO (Either () ByteString)+readBounded limit reader = loop 0 []+  where+    loop size chunks = do+      chunk <- HTTP.brRead reader+      if BS.null chunk+        then pure (Right (BS.concat (reverse chunks)))+        else+          if BS.length chunk > limit - size+            then pure (Left ())+            else loop (size + BS.length chunk) (chunk : chunks)++-- | One bounded raw row at a time; no decoded prefix escapes before framed EOF.+-- The newline delimiter is excluded from the byte bound; CR in CRLF is included.+-- Fragments avoid quadratic copying when a row arrives in many small chunks.+readJsonRowStream :: Int -> IO ByteString -> IO TransportOutcome+readJsonRowStream limit readChunk = loop 0 [] []+  where+    failure code = SubmissionUncertain (TransportError code (Just 200) Nothing)+    row bytes+      | BS.all (`elem` [9, 13, 32]) bytes = Right Nothing+      | BS.isPrefixOf "Code: " bytes = Left (classify JsonRows 200 Nothing bytes)+      | otherwise = do+          _ <- either (const (Left (failure UnexpectedResponse))) Right (Encoding.decodeUtf8' bytes)+          case eitherDecodeStrict' bytes :: Either String Object of+            Left _ -> Left (failure UnexpectedResponse)+            Right value -> Right (Just value)+    finish chunks rows = case row (BS.concat (reverse chunks)) of+      Left problem -> pure problem+      Right value -> pure (AcknowledgedSuccess (Rows (reverse (maybe rows (: rows) value))))+    loop size chunks rows = do+      chunk <- readChunk+      if BS.null chunk then finish chunks rows else consume size chunks rows chunk+    consume size chunks rows chunk =+      let (piece, rest) = BS.break (== 10) chunk+       in if BS.length piece > limit - size+            then pure (failure ResponseTooLarge)+            else+              if BS.null rest+                then loop (size + BS.length piece) (piece : chunks) rows+                else case row (BS.concat (reverse (piece : chunks))) of+                  Left problem -> pure problem+                  Right value ->+                    let nextRows = maybe rows (: rows) value+                        remaining = BS.tail rest+                     in if BS.null remaining then loop 0 [] nextRows else consume 0 [] nextRows remaining++classify :: RequestKind -> Int -> Maybe ByteString -> ByteString -> TransportOutcome+classify kind status exceptionHeader body+  | status >= 300 && status < 400 = ObservedServerFailure (TransportError RedirectRejected (Just status) serverCode)+  | status /= 200 || isJust exceptionHeader || isJust bodyCode = ObservedServerFailure (TransportError ServerRejected (Just status) serverCode)+  | kind == Command = if BS.all (`elem` [9, 10, 13, 32]) body then AcknowledgedSuccess CommandAccepted else unexpected+  | otherwise = case traverse (eitherDecodeStrict' :: ByteString -> Either String Object) (filter (not . BS.all (`elem` [9, 13, 32])) (Char8.lines body)) of+      Left _ -> unexpected+      Right rows -> AcknowledgedSuccess (Rows rows)+  where+    headerCode = exceptionHeader >>= readMaybe . Char8.unpack+    bodyCode = Char8.stripPrefix "Code: " body >>= readMaybe . Char8.unpack . Char8.takeWhile (\c -> c >= '0' && c <= '9')+    serverCode = case headerCode of Just code -> Just code; Nothing -> bodyCode+    unexpected = SubmissionUncertain (TransportError UnexpectedResponse (Just status) serverCode)++-- Probe on exactly the new owned connection, before returning it to the pool.+-- dummyManaged supplies no cleanup/replay; the factory owns this socket and closes+-- it on failure. The bounded response parser is the same as for ordinary requests.+probeConnection :: TargetConfig -> Internal.Connection -> QueryId -> ByteString -> IO TransportOutcome+probeConnection (TargetConfig _ (DirectEndpoint endpoint) _ _ _ (Credentials user password) (HttpLimits micros maxBytes)) connection (QueryId identity) sql = do+  result <- try @SomeException $ timeout micros $ do+    request <- makeRequest endpoint user password micros identity JsonRows sql+    continuation <- Internal.requestBuilder request connection+    bracket+      (Internal.getResponse (Just (Internal.MaxHeaderLength 4096)) (Just (Internal.MaxNumberHeaders 64)) (Just micros) request (Internal.dummyManaged connection) continuation)+      HTTP.responseClose+      ( \response -> do+          body <- readBounded maxBytes (HTTP.responseBody response)+          pure $ case body of+            Left () -> SubmissionUncertain (safeError ResponseTooLarge)+            Right bytes -> classify JsonRows (statusCode (HTTP.responseStatus response)) (lookup "X-ClickHouse-Exception-Code" (HTTP.responseHeaders response)) bytes+      )+  case result of+    Left exception | isJust (fromException exception :: Maybe SomeAsyncException) -> throwIO exception+    Left exception | Just (AdmissionFailure failure) <- fromException exception -> pure (SubmissionUncertain failure)+    Left _ -> pure (SubmissionUncertain (safeError ResponseInterrupted))+    Right Nothing -> pure (SubmissionUncertain (safeError DeadlineExpired))+    Right (Just outcome) -> pure outcome++makeRequest :: Text -> ByteString -> ByteString -> Int -> Text -> RequestKind -> ByteString -> IO HTTP.Request+makeRequest endpoint user password micros identity kind sql = do+  base <- HTTP.parseRequest (Text.unpack endpoint)+  let request =+        HTTP.applyBasicAuth user password $+          HTTP.setQueryString+            ( [ ("query_id", Just (Encoding.encodeUtf8 identity)),+                ("async_insert", Just "0"),+                ("mutations_sync", Just "1"),+                ("alter_sync", Just "1"),+                ("fsync_metadata", Just "1"),+                ("wait_end_of_query", Just "1"),+                ("default_format", Just "JSONEachRow")+              ]+                ++ if kind /= Command+                  then+                    [ ("max_rows_to_read", Just "0"),+                      ("max_bytes_to_read", Just "0"),+                      ("max_result_rows", Just "0"),+                      ("max_result_bytes", Just "0"),+                      ("read_overflow_mode", Just "throw"),+                      ("result_overflow_mode", Just "throw"),+                      ("timeout_overflow_mode", Just "throw"),+                      ("sort_overflow_mode", Just "throw")+                    ]+                  else []+            )+            base+              { HTTP.method = "POST",+                HTTP.requestBody = HTTP.RequestBodyBS sql,+                HTTP.redirectCount = 0,+                HTTP.checkResponse = \_ _ -> pure (),+                HTTP.responseTimeout = HTTP.responseTimeoutMicro micros,+                HTTP.decompress = const False,+                HTTP.requestHeaders = [("Content-Type", "text/plain; charset=utf-8"), ("Accept-Encoding", "identity")]+              }+  pure request++registered :: IORef [ThreadId] -> IO a -> IO a+registered threads action = mask $ \restore -> do+  identity <- myThreadId+  atomicModifyIORef' threads (\old -> (identity : old, ()))+  restore action `finally` atomicModifyIORef' threads (\old -> (filter (/= identity) old, ()))
+ src/ClickHouseMigrate/Internal/Identity.hs view
@@ -0,0 +1,23 @@+-- | Fresh client identities exist before any request is submitted.+module ClickHouseMigrate.Internal.Identity (freshUuid, diagnosticTime) where++import ClickHouseMigrate.Prelude+import Data.Bits ((.&.), (.|.))+import Data.ByteString qualified as BS+import Data.Text qualified as Text+import Data.Time (defaultTimeLocale, formatTime, getCurrentTime)+import Numeric (showHex)+import System.Entropy (getEntropy)++freshUuid :: IO Text+freshUuid = do+  bytes <- BS.unpack <$> getEntropy 16+  let normalized = [if i == 6 then (b .&. 15) .|. 64 else if i == 8 then (b .&. 63) .|. 128 else b | (i, b) <- zip [0 :: Int ..] bytes]+      hex = concatMap (\b -> let value = showHex b "" in replicate (2 - length value) '0' ++ value) normalized+  if length hex /= 32+    then fail "Insufficient identity entropy"+    else+      pure (Text.pack (take 8 hex ++ "-" ++ take 4 (drop 8 hex) ++ "-" ++ take 4 (drop 12 hex) ++ "-" ++ take 4 (drop 16 hex) ++ "-" ++ drop 20 hex))++diagnosticTime :: IO Text+diagnosticTime = Text.pack . formatTime defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" <$> getCurrentTime
+ src/ClickHouseMigrate/Internal/LedgerCodec.hs view
@@ -0,0 +1,316 @@+-- | Strict bounded version-one event codec; values are JSON data, never SQL.+module ClickHouseMigrate.Internal.LedgerCodec+  ( maxEventBytes,+    encodeEvent,+    decodeEvent,+    eventValue,+    definitionValue,+    definitions,+    validateEvent,+    eventKind,+    encodeValue,+    definitionsDigest,+    contextFor,+  )+where++import ClickHouseMigrate.Definition qualified as Def+import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.LedgerTypes+import ClickHouseMigrate.Internal.LedgerValidation (auditContext, ledgerPositionText, parseLedgerPosition)+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target qualified as Target+import Data.Aeson (Value (..), eitherDecodeStrict', object, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.Aeson.Types (Parser, parseEither, withObject, (.:))+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as Lazy+import Data.List (sortOn)+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Time (UTCTime, defaultTimeLocale, parseTimeM)++encodeValue :: Value -> ByteString+encodeValue = \case+  Object members -> "{" <> BS.intercalate "," [encodeValue (String (Key.toText key)) <> ":" <> encodeValue value | (key, value) <- sortOn (Key.toText . fst) (KM.toList members)] <> "}"+  Array values -> "[" <> BS.intercalate "," (foldr (\value rest -> encodeValue value : rest) [] values) <> "]"+  scalar -> Lazy.toStrict (Aeson.encode scalar)++maxEventBytes :: Int+maxEventBytes = 65536++encodeEvent :: LedgerEvent -> ByteString+encodeEvent = encodeValue . eventValue++eventValue :: LedgerEvent -> Value+eventValue (LedgerEvent sequenceNumber event ledger run attempt time audit runner context body) =+  object+    [ "version" .= (1 :: Int),+      "sequence" .= sequenceNumber,+      "eventId" .= event,+      "ledgerId" .= Target.ledgerIdText ledger,+      "runId" .= run,+      "attemptId" .= attempt,+      "occurredAt" .= time,+      "audit" .= auditValue audit,+      "runnerVersion" .= runner,+      "context" .= fmap contextValue context,+      "kind" .= eventKind body,+      "body" .= bodyValue body+    ]++eventKind :: EventBody -> Text+eventKind = \case+  Initialize _ -> "initialized"+  StartMigration _ -> "migration-started"+  StartStep _ _ -> "step-started"+  SucceedStep _ _ -> "step-succeeded"+  FailStep _ _ _ -> "execution-failed"+  CompleteMigration -> "migration-completed"+  AuthorizeRecovery _ -> "recovery-authorized"+  AcceptCompletion _ _ -> "step-accepted"+  StartAdoption _ -> "adoption-started"+  AdoptEntry _ _ -> "adoption-entry"+  CompleteAdoption _ -> "adoption-completed"++bodyValue :: EventBody -> Value+bodyValue = \case+  Initialize target -> object ["target" .= Target.targetKeyText target, "schemaVersion" .= (1 :: Int), "checksumFormat" .= ("sha256-v1" :: Text), "policy" .= Encoding.decodeUtf8 Plan.completionPolicy]+  StartMigration definition -> definitionValue definition+  StartStep step query -> stepValue step query+  SucceedStep step query -> stepValue step query+  FailStep step query (FailureEvidence code status server) -> object ["step" .= step, "queryId" .= query, "code" .= code, "httpStatus" .= status, "serverCode" .= server]+  CompleteMigration -> object []+  AuthorizeRecovery (Recovery action expected previous next (QuiescenceEvidence target audit)) ->+    object+      [ "action" .= recoveryName action,+        "step" .= recoveryStep action,+        "expected" .= ledgerPositionText expected,+        "previousAttempt" .= previous,+        "nextAttempt" .= next,+        "target" .= Target.targetKeyText target,+        "quiescence" .= auditValue audit+      ]+  AcceptCompletion step evidence -> object ["step" .= step, "evidence" .= evidenceValue evidence]+  StartAdoption (AdoptionBatch source snapshot mapping definitionsHash count) ->+    object+      ["source" .= source, "snapshot" .= snapshot, "mapping" .= mapping, "definitions" .= definitionsHash, "count" .= count]+  AdoptEntry definition evidence -> object ["definition" .= definitionValue definition, "evidence" .= evidenceValue evidence]+  CompleteAdoption digest -> object ["digest" .= digest]++stepValue :: Int -> Text -> Value+stepValue step query = object ["step" .= step, "queryId" .= query]++auditValue :: AuditContext -> Value+auditValue (AuditContext actor reason references) = object ["actor" .= actor, "reason" .= reason, "references" .= references]++evidenceValue :: Evidence -> Value+evidenceValue (Evidence level references) = object ["level" .= (case level of OperatorAttestation -> "operator-attestation" :: Text; PredecessorChecksum -> "predecessor-checksum"), "references" .= references]++definitionValue :: Definition -> Value+definitionValue (Definition ordinal componentIndex component dependencies migration checksum steps) =+  object+    [ "ordinal" .= ordinal,+      "componentIndex" .= componentIndex,+      "component" .= component,+      "dependencies" .= dependencies,+      "migration" .= migration,+      "checksum" .= checksum,+      "steps" .= [object ["name" .= name, "checksum" .= digest] | StepDefinition name digest <- steps]+    ]++contextValue :: EventContext -> Value+contextValue (EventContext component migration checksum step) =+  object+    [ "component" .= component,+      "migration" .= migration,+      "checksum" .= checksum,+      "step" .= fmap (\(name, digest) -> object ["name" .= name, "checksum" .= digest]) step+    ]++contextFor :: Definition -> Maybe Int -> Either LedgerError EventContext+contextFor definition index =+  EventContext (component definition) (migration definition) (checksum definition) <$> case index of+    Nothing -> Right Nothing+    Just i | i >= 0 -> case drop i (steps definition) of+      StepDefinition name digest : _ -> Right (Just (name, digest))+      _ -> Left InvalidEvent+    _ -> Left InvalidEvent++parseContext :: Value -> Parser EventContext+parseContext = exact ["component", "migration", "checksum", "step"] $ \o ->+  EventContext <$> o .: "component" <*> o .: "migration" <*> o .: "checksum" <*> (o .: "step" >>= traverse (exact ["name", "checksum"] (\s -> (,) <$> s .: "name" <*> s .: "checksum")))++recoveryName :: RecoveryAction -> Text+recoveryName = \case Resume -> "resume"; RetryStep _ -> "retry-step"; AcceptStep _ -> "accept-step"; Finalize -> "finalize"++recoveryStep :: RecoveryAction -> Maybe Int+recoveryStep = \case RetryStep step -> Just step; AcceptStep step -> Just step; _ -> Nothing++definitions :: Plan.MigrationPlan -> [Definition]+definitions plan =+  zipWith+    bind+    [0 ..]+    [ (index, component, migration)+    | migration <- NE.toList (Plan.inspectSchedule plan),+      (index, component) <- zip [0 ..] (NE.toList (Plan.inspectPlan plan)),+      Plan.componentViewName component == Def.migrationIdComponent (Plan.migrationId migration)+    ]+  where+    bind ordinal (index, component, migration) =+      Definition+        ordinal+        index+        (Def.componentNameText (Plan.componentViewName component))+        (map Def.componentNameText (Set.toAscList (Plan.componentDependencies component)))+        (Def.migrationNameText (Def.migrationIdName (Plan.migrationId migration)))+        (Plan.migrationChecksumHex (Plan.migrationChecksum migration))+        [StepDefinition (Def.stepNameText (Plan.stepViewName step)) (Plan.stepChecksumHex (Plan.stepChecksum step)) | step <- NE.toList (Plan.migrationSteps migration)]++definitionsDigest :: [Definition] -> Text+definitionsDigest values = Checksum.digestHex (Checksum.digest (encodeValue (Aeson.toJSON (map definitionValue values))))++decodeEvent :: ByteString -> Either LedgerError LedgerEvent+decodeEvent bytes+  | BS.length bytes > maxEventBytes = Left LedgerLimitExceeded+  | otherwise = do+      value <- either (const (Left InvalidEvent)) Right (eitherDecodeStrict' bytes)+      event <- either (const (Left InvalidEvent)) Right (parseEither parseEvent value)+      validateEvent event+      -- Our encoder supplies a canonical wire spelling. This also rejects duplicate+      -- keys (including escaped aliases), alternate number/string encodings and+      -- whitespace changes instead of discarding evidence during object decoding.+      if encodeEvent event == bytes then Right event else Left InvalidEvent++parseEvent :: Value -> Parser LedgerEvent+parseEvent = exact ["version", "sequence", "eventId", "ledgerId", "runId", "attemptId", "occurredAt", "audit", "runnerVersion", "context", "kind", "body"] $ \o -> do+  version <- o .: "version" :: Parser Int+  runner <- o .: "runnerVersion" :: Parser Text+  unless (version == 1 && not (Text.null runner) && Text.length runner <= 128) (fail "unsupported version")+  ledger <- o .: "ledgerId" >>= validated Target.ledgerId+  kind <- o .: "kind"+  body <- o .: "body" >>= parseBody kind+  LedgerEvent <$> o .: "sequence" <*> o .: "eventId" <*> pure ledger <*> o .: "runId" <*> o .: "attemptId" <*> o .: "occurredAt" <*> (o .: "audit" >>= parseAudit) <*> pure runner <*> (o .: "context" >>= traverse parseContext) <*> pure body++parseBody :: Text -> Value -> Parser EventBody+parseBody = \case+  "initialized" -> exact ["target", "schemaVersion", "checksumFormat", "policy"] $ \o -> do+    version <- o .: "schemaVersion" :: Parser Int+    format <- o .: "checksumFormat" :: Parser Text+    policy <- o .: "policy" :: Parser Text+    unless (version == 1 && format == "sha256-v1" && policy == Encoding.decodeUtf8 Plan.completionPolicy) (fail "unsupported format")+    Initialize <$> (o .: "target" >>= validated Target.targetKey)+  "migration-started" -> fmap StartMigration . parseDefinition+  "step-started" -> parseStep StartStep+  "step-succeeded" -> parseStep SucceedStep+  "execution-failed" -> exact ["step", "queryId", "code", "httpStatus", "serverCode"] $ \o -> FailStep <$> o .: "step" <*> o .: "queryId" <*> (FailureEvidence <$> o .: "code" <*> o .: "httpStatus" <*> o .: "serverCode")+  "migration-completed" -> exact [] (const (pure CompleteMigration))+  "recovery-authorized" -> exact ["action", "step", "expected", "previousAttempt", "nextAttempt", "target", "quiescence"] $ \o -> do+    name <- o .: "action" :: Parser Text+    step <- o .: "step" :: Parser (Maybe Int)+    action <- case (name, step) of+      ("resume", Nothing) -> pure Resume+      ("finalize", Nothing) -> pure Finalize+      ("retry-step", Just i) -> pure (RetryStep i)+      ("accept-step", Just i) -> pure (AcceptStep i)+      _ -> fail "invalid recovery"+    expected <- o .: "expected" >>= validated parseLedgerPosition+    target <- o .: "target" >>= validated Target.targetKey+    audit <- o .: "quiescence" >>= parseAudit+    AuthorizeRecovery <$> (Recovery action expected <$> o .: "previousAttempt" <*> o .: "nextAttempt" <*> pure (QuiescenceEvidence target audit))+  "step-accepted" -> exact ["step", "evidence"] $ \o -> AcceptCompletion <$> o .: "step" <*> (o .: "evidence" >>= parseEvidence)+  "adoption-started" -> exact ["source", "snapshot", "mapping", "definitions", "count"] $ \o -> StartAdoption <$> (AdoptionBatch <$> o .: "source" <*> o .: "snapshot" <*> o .: "mapping" <*> o .: "definitions" <*> o .: "count")+  "adoption-entry" -> exact ["definition", "evidence"] $ \o -> AdoptEntry <$> (o .: "definition" >>= parseDefinition) <*> (o .: "evidence" >>= parseEvidence)+  "adoption-completed" -> exact ["digest"] $ \o -> CompleteAdoption <$> o .: "digest"+  _ -> const (fail "unknown kind")++parseStep :: (Int -> Text -> EventBody) -> Value -> Parser EventBody+parseStep construct = exact ["step", "queryId"] $ \o -> construct <$> o .: "step" <*> o .: "queryId"++parseAudit :: Value -> Parser AuditContext+parseAudit = exact ["actor", "reason", "references"] $ \o -> do+  actor <- o .: "actor"+  reason <- o .: "reason"+  references <- o .: "references"+  validated (\() -> auditContext actor reason references) ()++parseEvidence :: Value -> Parser Evidence+parseEvidence = exact ["level", "references"] $ \o -> do+  level <- o .: "level" :: Parser Text+  Evidence <$> (case level of "operator-attestation" -> pure OperatorAttestation; "predecessor-checksum" -> pure PredecessorChecksum; _ -> fail "unknown evidence") <*> o .: "references"++parseDefinition :: Value -> Parser Definition+parseDefinition = exact ["ordinal", "componentIndex", "component", "dependencies", "migration", "checksum", "steps"] $ \o ->+  Definition <$> o .: "ordinal" <*> o .: "componentIndex" <*> o .: "component" <*> o .: "dependencies" <*> o .: "migration" <*> o .: "checksum" <*> (o .: "steps" >>= mapM (exact ["name", "checksum"] (\s -> StepDefinition <$> s .: "name" <*> s .: "checksum")))++exact :: [Text] -> (Aeson.Object -> Parser a) -> Value -> Parser a+exact keys action = withObject "ledger object" $ \o -> do+  unless (Set.fromList (map Key.toText (KM.keys o)) == Set.fromList keys) (fail "invalid fields")+  action o++validated :: (a -> Either e b) -> a -> Parser b+validated f = either (const (fail "invalid value")) pure . f++validateEvent :: LedgerEvent -> Either LedgerError ()+validateEvent event@(LedgerEvent sequenceNumber eventId ledger run attempt occurred audit runner context body) = do+  require (sequenceNumber > 0 && BS.length (encodeEvent event) <= maxEventBytes && not (Text.null runner) && Text.length runner <= 128)+  uuid eventId+  uuid run+  mapM_ uuid attempt+  _ <- auditContext (actor audit) (reason audit) (references audit)+  require (isJust (parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" (Text.unpack occurred) :: Maybe UTCTime))+  mapM_+    ( \(EventContext component migration checksum step) -> do+        void (either (const (Left InvalidEvent)) Right (Def.componentName component))+        void (either (const (Left InvalidEvent)) Right (Def.migrationName migration))+        checkHash checksum+        mapM_ (\(name, digest) -> void (either (const (Left InvalidEvent)) Right (Def.stepName name)) >> checkHash digest) step+    )+    context+  case body of+    Initialize _ -> require (isNothing attempt && sequenceNumber == 1 && isNothing context)+    StartMigration definition -> needAttempt >> checkDefinition definition >> (contextFor definition Nothing >>= require . (== context) . Just)+    StartStep step query -> needAttempt >> checkStep step query+    SucceedStep step query -> needAttempt >> checkStep step query+    FailStep step query (FailureEvidence code status server) -> needAttempt >> checkStep step query >> require (code `elem` ["ServerRejected", "UnexpectedResponse", "RedirectRejected"] && maybe True (\n -> n >= 100 && n <= 599) status && maybe True (>= 0) server)+    CompleteMigration -> needAttempt+    AuthorizeRecovery (Recovery action (LedgerPosition expected _ previousEvent) previous next (QuiescenceEvidence _ evidenceAudit)) -> do+      needAttempt+      uuid previous+      uuid next+      uuid previousEvent+      require (ledger == expected && previous /= next && all (>= 0) (recoveryStep action) && not (null (references evidenceAudit)))+      _ <- auditContext (actor evidenceAudit) (reason evidenceAudit) (references evidenceAudit)+      pure ()+    AcceptCompletion step evidence -> needAttempt >> require (step >= 0) >> checkEvidence evidence+    StartAdoption (AdoptionBatch source snapshot mapping definition count) -> do+      require (isNothing attempt && isNothing context && not (Text.null (Text.strip source)) && Text.length source <= 2048 && count > 0)+      mapM_ checkHash [snapshot, mapping, definition]+    AdoptEntry definition evidence -> require (isNothing attempt) >> checkDefinition definition >> checkEvidence evidence >> (contextFor definition Nothing >>= require . (== context) . Just)+    CompleteAdoption digest -> require (isNothing attempt && isNothing context) >> checkHash digest+  where+    require True = Right (); require False = Left InvalidEvent+    uuid = either (const (Left InvalidEvent)) (const (Right ())) . Target.ledgerId+    needAttempt = require (isJust attempt && isJust context)+    checkStep step query = require (step >= 0) >> either (const (Left InvalidEvent)) (const (Right ())) (Target.queryId query)+    checkHash digest = require (Text.length digest == 64 && Text.all (\c -> (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) digest)+    checkEvidence (Evidence _ refs) = do+      require (not (null refs))+      void (auditContext "evidence" "evidence" refs)+    checkDefinition (Definition ordinal index component dependencies migration checksum steps) = do+      require (ordinal >= 0 && index >= 0 && not (null steps) && dependencies == Set.toAscList (Set.fromList dependencies) && component `notElem` dependencies)+      void (either (const (Left InvalidEvent)) Right (Def.componentName component))+      void (either (const (Left InvalidEvent)) Right (Def.migrationName migration))+      mapM_ (\dependency -> void (either (const (Left InvalidEvent)) Right (Def.componentName dependency))) dependencies+      checkHash checksum+      require (length steps == Set.size (Set.fromList [name | StepDefinition name _ <- steps]))+      mapM_ (\(StepDefinition name digest) -> void (either (const (Left InvalidEvent)) Right (Def.stepName name)) >> checkHash digest) steps+    actor (AuditContext value _ _) = value+    reason (AuditContext _ value _) = value+    references (AuditContext _ _ value) = value
+ src/ClickHouseMigrate/Internal/LedgerSchema.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE MultilineStrings #-}++-- | Exact local v1 physical shape; never infer uniqueness from the sorting key.+module ClickHouseMigrate.Internal.LedgerSchema where++import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (LedgerNamespace, ledgerNamespaceText)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding++quoteIdentifier :: Text -> Text+quoteIdentifier value = "`" <> Text.replace "`" "``" value <> "`"++qualifiedTable :: LedgerNamespace -> Text+qualifiedTable namespace = quoteIdentifier (ledgerNamespaceText namespace) <> ".`events`"++createDatabase :: LedgerNamespace -> ByteString+createDatabase namespace = Encoding.encodeUtf8 ("CREATE DATABASE IF NOT EXISTS " <> quoteIdentifier (ledgerNamespaceText namespace) <> " ENGINE=Atomic")++createTable :: LedgerNamespace -> ByteString+createTable namespace = Encoding.encodeUtf8 ("CREATE TABLE IF NOT EXISTS " <> qualifiedTable namespace <> " " <> columnsSql <> " ENGINE=MergeTree ORDER BY sequence SETTINGS fsync_after_insert=1, fsync_part_directory=1, storage_policy='default'")++columnsSql :: Text+columnsSql = "(sequence UInt64, event_id UUID, ledger_id UUID, event String)"++expectedCreate :: LedgerNamespace -> Text+expectedCreate namespace = "CREATE TABLE " <> ledgerNamespaceText namespace <> ".events (`sequence` UInt64, `event_id` UUID, `ledger_id` UUID, `event` String) ENGINE = " <> expectedEngine++expectedEngine :: Text+expectedEngine = "MergeTree ORDER BY sequence SETTINGS fsync_after_insert = 1, fsync_part_directory = 1, storage_policy = 'default', index_granularity = 8192"++-- Only validated ASCII namespace names enter literals; event/audit data are JSON.+namespaceLiteral :: LedgerNamespace -> Text+namespaceLiteral = (\value -> "'" <> value <> "'") . ledgerNamespaceText++databaseQuery :: LedgerNamespace -> ByteString+databaseQuery namespace = Encoding.encodeUtf8 ("SELECT name, engine FROM system.databases WHERE name=" <> namespaceLiteral namespace)++tableQuery :: LedgerNamespace -> ByteString+tableQuery namespace = Encoding.encodeUtf8 ("SELECT name, engine, engine_full, sorting_key, primary_key, partition_key, sampling_key, create_table_query FROM system.tables WHERE database=" <> namespaceLiteral namespace <> " ORDER BY name")++readQuery :: LedgerNamespace -> ByteString+readQuery namespace = Encoding.encodeUtf8 ("SELECT toString(sequence) AS sequence, toString(event_id) AS event_id, toString(ledger_id) AS ledger_id, event FROM " <> qualifiedTable namespace <> " ORDER BY sequence, event_id")
+ src/ClickHouseMigrate/Internal/LedgerStore.hs view
@@ -0,0 +1,261 @@+-- | Guarded v1 storage. Every append is frozen, submitted once, then acknowledged+-- or made blocking. No reconciliation path writes or resumes an operation.+module ClickHouseMigrate.Internal.LedgerStore where++import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.Exclusive (checkGuard)+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessError, Guard)+import ClickHouseMigrate.Internal.Identity (diagnosticTime, freshUuid)+import ClickHouseMigrate.Internal.LedgerCodec (decodeEvent, encodeEvent, encodeValue)+import ClickHouseMigrate.Internal.LedgerSchema qualified as Schema+import ClickHouseMigrate.Internal.LedgerTypes+import ClickHouseMigrate.Internal.Projection (appendProjection, project)+import ClickHouseMigrate.Internal.TargetTypes (TargetConfig (..))+import ClickHouseMigrate.Internal.Transport+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target hiding (ledgerId)+import Control.Concurrent.MVar (MVar, newMVar, withMVar)+import Control.Exception (Exception, throwIO, try)+import Data.Aeson (FromJSON, Key, Object, Result (..), fromJSON, object, (.=))+import Data.Aeson.KeyMap qualified as KM+import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Text.Read (readMaybe)++data StoreError = LedgerProblem !LedgerError | AccessProblem !AccessError | RequestProblem !TransportOutcome !(Maybe AppendReceipt) | InitializationProblem !LedgerId !Text !QueryId !TransportOutcome+  deriving stock (Eq, Show)++data StoreAbort = StoreAbort !StoreError deriving stock (Show)++instance Exception StoreAbort++data Observation = Missing | EmptyNamespace | History !Projection deriving stock (Eq, Show)++data LedgerStore = LedgerStore !Guard !Transport !TargetKey !LedgerNamespace !(IORef (Either StoreError Projection)) !(MVar ())++storeResult :: IO a -> IO (Either StoreError a)+storeResult action = either (\(StoreAbort failure) -> Left failure) Right <$> try action++abort :: StoreError -> IO a+abort = throwIO . StoreAbort++checked :: Either LedgerError a -> IO a+checked = either (abort . LedgerProblem) pure++newQueryId :: IO QueryId+newQueryId = freshUuid >>= either (const (abort (LedgerProblem InvalidEvent))) pure . queryId . ("chm-" <>)++checkAccess :: Guard -> IO ()+checkAccess guard = checkGuard guard >>= either (abort . AccessProblem) pure++readRows :: Maybe Guard -> Transport -> ByteString -> IO [Object]+readRows = readRowsKind JsonRows++readRowsKind :: RequestKind -> Maybe Guard -> Transport -> ByteString -> IO [Object]+readRowsKind kind guard transport sql = do+  mapM_ checkAccess guard+  identity <- newQueryId+  outcome <- request transport identity kind sql+  case outcome of+    AcknowledgedSuccess (Rows rows) -> pure rows+    _ -> abort (RequestProblem outcome Nothing)++member :: (FromJSON a) => Key -> Object -> IO a+member key row = case KM.lookup key row of+  Just value -> case fromJSON value of Success result -> pure result; Error _ -> invalid+  _ -> invalid+  where+    invalid = abort (LedgerProblem InvalidLedgerSchema)++keys :: [Key] -> Object -> IO ()+keys expected row = unless (Set.fromList expected == Set.fromList (KM.keys row)) (abort (LedgerProblem InvalidLedgerSchema))++-- Shape state is separate from event projection to avoid inventing an empty+-- ledger identity. Return False for absent table after checking every object.+shape :: Maybe Guard -> Transport -> LedgerNamespace -> IO (Bool, Bool)+shape guard transport namespace = do+  checkLocalStorage guard transport+  settings <- readRows guard transport "SELECT toUInt8(getSetting('fsync_metadata')) AS fsync_metadata"+  case settings of+    [setting] -> do+      keys ["fsync_metadata"] setting+      effective <- member "fsync_metadata" setting :: IO Int+      unless (effective == 1) invalid+    _ -> invalid+  databases <- readRows guard transport (Schema.databaseQuery namespace)+  case databases of+    [] -> pure (False, False)+    [row] -> do+      keys ["name", "engine"] row+      name <- member "name" row+      engine <- member "engine" row :: IO Text+      unless (name == ledgerNamespaceText namespace && engine == "Atomic") invalid+      tables <- readRows guard transport (Schema.tableQuery namespace)+      case tables of+        [] -> pure (True, False)+        [table] -> do+          keys ["name", "engine", "engine_full", "sorting_key", "primary_key", "partition_key", "sampling_key", "create_table_query"] table+          actual <- mapM (`member` table) ["name", "engine", "engine_full", "sorting_key", "primary_key", "partition_key", "sampling_key", "create_table_query"] :: IO [Text]+          let base = ["events", "MergeTree", Schema.expectedEngine, "sequence", "sequence", "", ""]+              quotedCreate = Text.replace ("CREATE TABLE " <> ledgerNamespaceText namespace <> ".") ("CREATE TABLE " <> Schema.quoteIdentifier (ledgerNamespaceText namespace) <> ".") (Schema.expectedCreate namespace)+          unless (actual == base ++ [Schema.expectedCreate namespace] || actual == base ++ [quotedCreate]) invalid+          pure (True, True)+        _ -> invalid+    _ -> invalid+  where+    invalid = abort (LedgerProblem InvalidLedgerSchema)++-- Fsync-qualified v1 storage is the explicitly selected default policy backed by+-- healthy writable local disks. A same-named policy can otherwise point at object+-- storage, for which the native process-crash qualification does not apply.+checkLocalStorage :: Maybe Guard -> Transport -> IO ()+checkLocalStorage guard transport = do+  policies <- readRows guard transport "SELECT disks FROM system.storage_policies WHERE policy_name='default'"+  names <- concat <$> mapM (\row -> keys ["disks"] row >> (member "disks" row :: IO [Text])) policies+  unless (not (null names) && length names == Set.size (Set.fromList names)) invalid+  disks <- readRows guard transport "SELECT name, type, is_remote, is_read_only, is_broken FROM system.disks"+  decoded <-+    mapM+      ( \row -> do+          keys ["name", "type", "is_remote", "is_read_only", "is_broken"] row+          name <- member "name" row+          kind <- member "type" row :: IO Text+          flags <- mapM (`member` row) ["is_remote", "is_read_only", "is_broken"] :: IO [Int]+          pure (name, kind, flags)+      )+      disks+  mapM_+    ( \name -> case [(kind, flags) | (actual, kind, flags) <- decoded, actual == name] of+        [("Local", [0, 0, 0])] -> pure ()+        _ -> invalid+    )+    names+  where+    invalid = abort (LedgerProblem UnsupportedLedgerStorage)++load :: Maybe Guard -> Transport -> TargetConfig -> IO Observation+load guard transport (TargetConfig target _ _ namespace binding _ _) = do+  (database, table) <- shape guard transport namespace+  if not database+    then pure Missing+    else+      if not table+        then pure EmptyNamespace+        else do+          rows <- readRowsKind JsonRowStream guard transport (Schema.readQuery namespace)+          if null rows+            then pure EmptyNamespace+            else do+              events <- mapM decodeRow rows+              projection <- checked (project target events)+              case binding of+                ExistingLedger expected -> let LedgerPosition actual _ _ = position projection in unless (expected == actual) (abort (LedgerProblem LedgerIdentityMismatch))+                FirstUse -> pure ()+              pure (History projection)++decodeRow :: Object -> IO LedgerEvent+decodeRow row = do+  keys ["sequence", "event_id", "ledger_id", "event"] row+  encoded <- member "event" row+  event <- checked (decodeEvent (Encoding.encodeUtf8 encoded))+  number <- member "sequence" row :: IO Text+  identity <- member "event_id" row+  ledger <- member "ledger_id" row+  let LedgerEvent sequenceNumber eventIdentity ledgerIdentity _ _ _ _ _ _ _ = event+  unless (readMaybe (Text.unpack number) == Just sequenceNumber && identity == eventIdentity && ledger == ledgerIdText ledgerIdentity) (abort (LedgerProblem InvalidEvent))+  pure event++openStore :: Guard -> Transport -> TargetConfig -> Maybe InitializationAuthorization -> QuiescenceEvidence -> Text -> IO LedgerStore+openStore guard transport config@(TargetConfig target _ _ namespace binding _ _) authorization (QuiescenceEvidence quietTarget _) run = do+  checkAccess guard+  case authorization of+    Just (InitializationAuthorization authorized identity _) -> do+      unless (authorized == target) (abort (LedgerProblem InitializationNotAuthorized))+      case binding of ExistingLedger expected -> unless (expected == identity) (abort (LedgerProblem LedgerIdentityMismatch)); FirstUse -> pure ()+    Nothing -> pure ()+  unless (quietTarget == target) (abort (LedgerProblem OutstandingRequests))+  observation <- load (Just guard) transport config+  projection <- case observation of+    History state -> do+      case binding of+        ExistingLedger _ -> pure ()+        FirstUse -> case authorization of+          Just (InitializationAuthorization expected identity _) -> do+            let LedgerPosition actual _ _ = position state+            unless (target == expected && identity == actual) (abort (LedgerProblem LedgerIdentityMismatch))+          _ -> abort (LedgerProblem InitializationNotAuthorized)+      pure state+    _ -> case (binding, authorization) of+      (FirstUse, Just (InitializationAuthorization expected ledger authAudit)) | expected == target -> initialize guard transport config ledger authAudit run+      _ -> abort (LedgerProblem InitializationNotAuthorized)+  state <- newIORef (Right projection)+  gate <- newMVar ()+  pure (LedgerStore guard transport target namespace state gate)++initialize :: Guard -> Transport -> TargetConfig -> LedgerId -> AuditContext -> Text -> IO Projection+initialize guard transport (TargetConfig target _ _ namespace _ _ _) ledger audit run = do+  identity <- freshUuid+  time <- diagnosticTime+  let first = LedgerEvent 1 identity ledger run Nothing time audit "0.1.0.0" Nothing (Initialize target)+  _ <- checked (project target [first])+  ddl "database" (Schema.createDatabase namespace)+  _ <- shape (Just guard) transport namespace+  ddl "table" (Schema.createTable namespace)+  (_, exists) <- shape (Just guard) transport namespace+  unless exists (abort (LedgerProblem InvalidLedgerSchema))+  -- Even under exclusivity do not write over preexisting nonempty history.+  rows <- readRowsKind JsonRowStream (Just guard) transport (Schema.readQuery namespace)+  unless (null rows) (abort (LedgerProblem MissingInitialization))+  appendOnce guard transport target namespace first+  checked (project target [first])+  where+    ddl stage sql = do+      checkAccess guard+      identity <- newQueryId+      outcome <- request transport identity Command sql+      case outcome of AcknowledgedSuccess CommandAccepted -> pure (); _ -> abort (InitializationProblem ledger stage identity outcome)++appendOnce :: Guard -> Transport -> TargetKey -> LedgerNamespace -> LedgerEvent -> IO ()+appendOnce guard transport target namespace event@(LedgerEvent sequenceNumber identity ledger _ _ _ _ _ _ _) = do+  checkAccess guard+  query <- newQueryId+  let payload = object ["sequence" .= sequenceNumber, "event_id" .= identity, "ledger_id" .= ledgerIdText ledger, "event" .= Encoding.decodeUtf8 (encodeEvent event)]+      sql = Encoding.encodeUtf8 ("INSERT INTO " <> Schema.qualifiedTable namespace <> " FORMAT JSONEachRow\n") <> encodeValue payload <> "\n"+      receipt = AppendReceipt target ledger sequenceNumber identity (Checksum.digestHex (Checksum.digest (encodeEvent event))) query+  outcome <- request transport query Command sql+  case outcome of AcknowledgedSuccess CommandAccepted -> pure (); _ -> abort (RequestProblem outcome (Just receipt))++storeProjection :: LedgerStore -> IO Projection+storeProjection (LedgerStore _ _ _ _ state _) = readIORef state >>= either abort pure++append :: LedgerStore -> LedgerEvent -> IO Projection+append store@(LedgerStore guard transport target namespace state gate) event = withMVar gate $ \() -> do+  previous <- storeProjection store+  next <- checked (appendProjection previous event)+  result <- storeResult (appendOnce guard transport target namespace event)+  case result of+    Left failure -> writeIORef state (Left failure) >> abort failure+    Right () -> writeIORef state (Right next) >> pure next++-- A fresh guarded FULL read, including strict corruption validation. NotObserved+-- is evidence of this read only; it grants no authority to reuse a sequence.+reconcile :: Guard -> Transport -> TargetConfig -> AppendReceipt -> IO (Reconciliation, Observation)+reconcile guard transport config@(TargetConfig target _ _ _ _ _ _) (AppendReceipt expected ledger number identity digest _) = do+  unless (target == expected) (abort (LedgerProblem LedgerIdentityMismatch))+  observation <- load (Just guard) transport config+  let result = case observation of+        History projection ->+          let LedgerPosition actual _ _ = position projection+              matches = [event | event <- events projection, eventId event == identity || eventSequence event == number]+           in if actual /= ledger+                then Conflicting+                else case matches of+                  [] -> NotObserved+                  [event] | eventId event == identity && eventSequence event == number && Checksum.digestHex (Checksum.digest (encodeEvent event)) == digest -> PresentIdentical+                  _ -> Conflicting+        _ -> NotObserved+  pure (result, observation)+  where+    eventSequence (LedgerEvent value _ _ _ _ _ _ _ _ _) = value
+ src/ClickHouseMigrate/Internal/LedgerTypes.hs view
@@ -0,0 +1,123 @@+-- | Private ledger protocol. No unchecked construction is exported publicly.+module ClickHouseMigrate.Internal.LedgerTypes where++import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (LedgerId, QueryId, TargetKey)+import Data.Word (Word64)++-- Tokens contain observations, never authority to mutate.+data AttemptId = AttemptId !Text deriving stock (Eq, Ord, Show)++data AppendReceipt = AppendReceipt !TargetKey !LedgerId !Word64 !Text !Text !QueryId deriving stock (Eq, Show)++data Reconciliation = PresentIdentical | Conflicting | NotObserved deriving stock (Eq, Show)++data LedgerPosition = LedgerPosition !LedgerId !Word64 !Text deriving stock (Eq, Ord, Show)++data ExpectedLedgerState = ExpectAbsent !TargetKey | ExpectPosition !LedgerPosition deriving stock (Eq, Show)++data AuditContext = AuditContext !Text !Text ![Text] deriving stock (Eq, Show)++data QuiescenceEvidence = QuiescenceEvidence !TargetKey !AuditContext deriving stock (Eq, Show)++data InitializationAuthorization = InitializationAuthorization !TargetKey !LedgerId !AuditContext deriving stock (Eq, Show)++data EvidenceLevel = OperatorAttestation | PredecessorChecksum deriving stock (Eq, Ord, Show)++data Evidence = Evidence !EvidenceLevel ![Text] deriving stock (Eq, Show)++data LedgerError+  = InvalidPosition+  | InvalidAudit+  | InvalidEvidence+  | InvalidEvent+  | UnsupportedLedgerVersion+  | SequenceGap+  | SequenceConflict+  | EventIdentityConflict+  | LedgerIdentityMismatch+  | MissingInitialization+  | ImpossibleTransition+  | DefinitionMismatch+  | InvalidLedgerSchema+  | UnsupportedLedgerStorage+  | LedgerAbsent+  | LedgerUninitialized+  | InitializationNotAuthorized+  | StaleLedgerState+  | UnresolvedHistory+  | OutstandingRequests+  | LedgerLimitExceeded+  deriving stock (Eq, Ord, Show)++data Definition = Definition+  { ordinal :: !Int,+    componentIndex :: !Int,+    component :: !Text,+    dependencies :: ![Text],+    migration :: !Text,+    checksum :: !Text,+    steps :: ![StepDefinition]+  }+  deriving stock (Eq, Show)++data StepDefinition = StepDefinition !Text !Text deriving stock (Eq, Show)++data FailureEvidence = FailureEvidence !Text !(Maybe Int) !(Maybe Int) deriving stock (Eq, Show)++data RecoveryAction = Resume | RetryStep !Int | AcceptStep !Int | Finalize deriving stock (Eq, Show)++data Recovery = Recovery !RecoveryAction !LedgerPosition !Text !Text !QuiescenceEvidence deriving stock (Eq, Show)++data AdoptionBatch = AdoptionBatch !Text !Text !Text !Text !Int deriving stock (Eq, Show)++data EventBody+  = Initialize !TargetKey+  | StartMigration !Definition+  | StartStep !Int !Text+  | SucceedStep !Int !Text+  | FailStep !Int !Text !FailureEvidence+  | CompleteMigration+  | AuthorizeRecovery !Recovery+  | AcceptCompletion !Int !Evidence+  | StartAdoption !AdoptionBatch+  | AdoptEntry !Definition !Evidence+  | CompleteAdoption !Text+  deriving stock (Eq, Show)++data EventContext = EventContext !Text !Text !Text !(Maybe (Text, Text)) deriving stock (Eq, Show)++data LedgerEvent = LedgerEvent+  { sequence :: !Word64,+    eventId :: !Text,+    ledgerId :: !LedgerId,+    runId :: !Text,+    attemptId :: !(Maybe Text),+    occurredAt :: !Text,+    audit :: !AuditContext,+    runnerVersion :: !Text,+    context :: !(Maybe EventContext),+    body :: !EventBody+  }+  deriving stock (Eq, Show)++data StepState = NeverStarted | Started !Text | Succeeded !Text | Failed !Text !FailureEvidence | Accepted !Evidence+  deriving stock (Eq, Show)++data MigrationState = Applied | Adopted deriving stock (Eq, Show)++data RecordedMigration = RecordedMigration !Definition !MigrationState ![StepState] deriving stock (Eq, Show)++data ActiveMigration = ActiveMigration !Definition !Text ![StepState] !(Maybe RecoveryAction) deriving stock (Eq, Show)++data ActiveAdoption = ActiveAdoption !AdoptionBatch ![RecordedMigration] deriving stock (Eq, Show)++data Projection = Projection+  { position :: !LedgerPosition,+    target :: !TargetKey,+    completed :: ![RecordedMigration],+    active :: !(Maybe ActiveMigration),+    adoption :: !(Maybe ActiveAdoption),+    events :: ![LedgerEvent]+  }+  deriving stock (Eq, Show)
+ src/ClickHouseMigrate/Internal/LedgerValidation.hs view
@@ -0,0 +1,117 @@+-- | Validated ledger observations and explicit administrative assertions.+-- Assertions document caller evidence; they do not independently prove fencing.+module ClickHouseMigrate.Internal.LedgerValidation+  ( AttemptId,+    parseAttemptId,+    attemptIdText,+    LedgerPosition,+    ledgerPositionText,+    parseLedgerPosition,+    AppendReceipt,+    appendReceiptText,+    parseAppendReceipt,+    appendQueryId,+    ExpectedLedgerState (ExpectAbsent, ExpectPosition),+    AuditContext,+    auditContext,+    auditActor,+    auditReason,+    auditReferences,+    QuiescenceEvidence,+    attestQuiescence,+    quiescenceTarget,+    InitializationAuthorization,+    authorizeInitialization,+    EvidenceLevel (OperatorAttestation, PredecessorChecksum),+    LedgerError (..),+  )+where++import ClickHouseMigrate.Internal.LedgerTypes+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target qualified as Target+import Data.ByteString qualified as BS+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Word (Word64)+import Numeric (readHex, showHex)+import Text.Read (readMaybe)++ledgerPositionText :: LedgerPosition -> Text+ledgerPositionText (LedgerPosition identity sequenceNumber event) =+  "chm1:" <> Target.ledgerIdText identity <> ":" <> Text.pack (show sequenceNumber) <> ":" <> event++parseLedgerPosition :: Text -> Either LedgerError LedgerPosition+parseLedgerPosition encoded = case Text.splitOn ":" encoded of+  ["chm1", uuid, number, event] -> do+    identity <- either (const (Left InvalidPosition)) Right (Target.ledgerId uuid)+    _ <- either (const (Left InvalidPosition)) Right (Target.ledgerId event)+    sequenceNumber <- maybe (Left InvalidPosition) Right (readMaybe (Text.unpack number) :: Maybe Word64)+    let result = LedgerPosition identity sequenceNumber event+    if sequenceNumber > 0 && ledgerPositionText result == encoded then Right result else Left InvalidPosition+  _ -> Left InvalidPosition++auditContext :: Text -> Text -> [Text] -> Either LedgerError AuditContext+auditContext actor reason references+  | bounded 256 actor && bounded 4096 reason && length references <= 32 && all (bounded 2048) references = Right (AuditContext actor reason references)+  | otherwise = Left InvalidAudit+  where+    bounded limit value = not (Text.null (Text.strip value)) && Text.length value <= limit && not (Text.any (== '\0') value)++auditActor :: AuditContext -> Text+auditActor (AuditContext value _ _) = value++auditReason :: AuditContext -> Text+auditReason (AuditContext _ value _) = value++auditReferences :: AuditContext -> [Text]+auditReferences (AuditContext _ _ value) = value++-- | Caller attests that every predecessor has been fenced and ALL of its server+-- requests (including initialization DDL/ledger INSERTs) have ended. An empty+-- read, released OS lock, expired lease or absent process ID is insufficient.+attestQuiescence :: Target.TargetKey -> AuditContext -> Either LedgerError QuiescenceEvidence+attestQuiescence target audit@(AuditContext _ _ references)+  | null references = Left InvalidEvidence+  | otherwise = Right (QuiescenceEvidence target audit)++-- | A one-time target/UUID-bound first-use authorization. Persist this UUID with+-- an interrupted bootstrap; never generate a different one to resume it.+authorizeInitialization :: Target.TargetKey -> Target.LedgerId -> AuditContext -> InitializationAuthorization+authorizeInitialization = InitializationAuthorization++appendReceiptText :: AppendReceipt -> Text+appendReceiptText (AppendReceipt target ledger sequenceNumber event digest query) =+  Text.intercalate ":" ["chm-receipt1", hex (Encoding.encodeUtf8 (Target.targetKeyText target)), Target.ledgerIdText ledger, Text.pack (show sequenceNumber), event, digest, Target.queryIdText query]+  where+    hex = Text.pack . concatMap (\byte -> let value = showHex byte "" in replicate (2 - length value) '0' ++ value) . BS.unpack++parseAppendReceipt :: Text -> Either LedgerError AppendReceipt+parseAppendReceipt encoded = case Text.splitOn ":" encoded of+  ["chm-receipt1", targetHex, ledger, number, event, digest, query] | Text.length encoded <= 1024 -> do+    bytes <- unhex (Text.unpack targetHex)+    targetText <- either (const (Left InvalidPosition)) Right (Encoding.decodeUtf8' (BS.pack bytes))+    target <- either (const (Left InvalidPosition)) Right (Target.targetKey targetText)+    identity <- either (const (Left InvalidPosition)) Right (Target.ledgerId ledger)+    _ <- either (const (Left InvalidPosition)) Right (Target.ledgerId event)+    sequenceNumber <- maybe (Left InvalidPosition) Right (readMaybe (Text.unpack number) :: Maybe Word64)+    request <- either (const (Left InvalidPosition)) Right (Target.queryId query)+    let result = AppendReceipt target identity sequenceNumber event digest request+    if sequenceNumber > 0 && Text.length digest == 64 && Text.all (\c -> c >= '0' && c <= '9' || c >= 'a' && c <= 'f') digest && appendReceiptText result == encoded then Right result else Left InvalidPosition+  _ -> Left InvalidPosition+  where+    unhex [] = Right []+    unhex (a : b : rest) = case readHex [a, b] of [(byte, "")] -> (byte :) <$> unhex rest; _ -> Left InvalidPosition+    unhex _ = Left InvalidPosition++quiescenceTarget :: QuiescenceEvidence -> Target.TargetKey+quiescenceTarget (QuiescenceEvidence target _) = target++appendQueryId :: AppendReceipt -> Target.QueryId+appendQueryId (AppendReceipt _ _ _ _ _ query) = query++parseAttemptId :: Text -> Either LedgerError AttemptId+parseAttemptId value = either (const (Left InvalidPosition)) (const (Right (AttemptId value))) (Target.ledgerId value)++attemptIdText :: AttemptId -> Text+attemptIdText (AttemptId value) = value
+ src/ClickHouseMigrate/Internal/Preflight.hs view
@@ -0,0 +1,48 @@+-- | Force the complete definition/target envelope before contacting a server.+module ClickHouseMigrate.Internal.Preflight (preflight, validateRecoveryEvent, validateRecoveryCapacity) where++import ClickHouseMigrate.Internal.LedgerCodec (contextFor, definitions, encodeEvent, maxEventBytes, validateEvent)+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Reporting (stepReference, summaryDefinition)+import ClickHouseMigrate.Internal.SqlEnvelope (mentionsIdentifier, validateSql)+import ClickHouseMigrate.Internal.TargetTypes qualified as T+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Target (ledgerNamespaceText)+import Data.ByteString qualified as BS+import Data.List.NonEmpty qualified as NE+import Data.Word (Word64)++preflight :: T.TargetConfig -> L.AuditContext -> Plan.MigrationPlan -> Either R.RunError ()+preflight (T.TargetConfig target _ _ namespace _ _ _) audit plan = do+  let uuid = "00000000-0000-4000-8000-000000000000"+      envelope body context = L.LedgerEvent 1 uuid (T.LedgerId uuid) uuid (Just uuid) "2026-10-16T00:00:00.999999999999Z" audit "0.1.0.0" context body+      authored = [(migration, NE.toList (Plan.migrationSteps migration)) | migration <- NE.toList (Plan.inspectSchedule plan)]+      sources = zip (definitions plan) authored+  if BS.length (encodeEvent (envelope (L.Initialize target) Nothing)) > 61440 then Left R.AuditTooLarge else pure ()+  mapM_+    ( \(definition, (_, steps)) -> do+        context <- either (const (Left (R.DefinitionTooLarge (summaryDefinition definition)))) Right (contextFor definition Nothing)+        if BS.length (encodeEvent (envelope (L.StartMigration definition) (Just context))) > 61440 then Left (R.DefinitionTooLarge (summaryDefinition definition)) else pure ()+        mapM_+          ( \(index, step) -> case stepReference definition index of+              Nothing -> Left R.UnexpectedRuntimeFailure+              Just reference -> do+                either (const (Left (R.InvalidRuntimePlan reference))) Right (validateSql (Plan.stepSql step))+                mentioned <- either (const (Left (R.InvalidRuntimePlan reference))) Right (mentionsIdentifier (ledgerNamespaceText namespace) (Plan.stepSql step))+                if mentioned then Left (R.LedgerNamespaceInSql reference) else pure ()+          )+          (zip [0 ..] steps)+    )+    sources++-- Count the one canonical UTF-8/JSON encoding, including repeated evidence.+validateRecoveryEvent :: R.RecoveryEventKind -> L.LedgerEvent -> Either R.RunError ()+validateRecoveryEvent kind event+  | BS.length (encodeEvent event) > maxEventBytes = Left (R.RecoveryEventTooLarge kind)+  | otherwise = either (Left . R.LedgerFailed) Right (validateEvent event)++validateRecoveryCapacity :: Word64 -> Integer -> Either R.RunError ()+validateRecoveryCapacity current needed+  | needed < 0 || toInteger current + needed > toInteger (maxBound :: Word64) = Left (R.LedgerFailed L.LedgerLimitExceeded)+  | otherwise = Right ()
+ src/ClickHouseMigrate/Internal/Projection.hs view
@@ -0,0 +1,180 @@+-- | Pure immutable history fold. No timestamp or merge order chooses a winner.+module ClickHouseMigrate.Internal.Projection (project, appendProjection, comparePlan, unresolved, resolvedStep) where++import ClickHouseMigrate.Internal.LedgerCodec (contextFor, definitions, definitionsDigest, validateEvent)+import ClickHouseMigrate.Internal.LedgerTypes+import ClickHouseMigrate.Plan (MigrationPlan)+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (TargetKey)+import Control.Monad (foldM)+import Data.List (findIndex)+import Data.Map.Strict qualified as Map++project :: TargetKey -> [LedgerEvent] -> Either LedgerError Projection+project _ [] = Left LedgerUninitialized+project target rows = do+  ordered <- deduplicate rows+  case ordered of+    first@(LedgerEvent 1 event ledger _ Nothing _ _ _ Nothing (Initialize actual)) : rest+      | target == actual -> do+          validateEvent first+          foldM appendProjection (Projection (LedgerPosition ledger 1 event) target [] Nothing Nothing [first]) rest+      | otherwise -> Left LedgerIdentityMismatch+    _ -> Left MissingInitialization++-- Equality is byte-equivalent because the strict codec admits only canonical v1+-- bytes. Detect identity conflicts before sorting/collapsing, never take a winner.+deduplicate :: [LedgerEvent] -> Either LedgerError [LedgerEvent]+deduplicate rows = do+  (_, bySequence) <- foldM add (Map.empty, Map.empty) rows+  pure (map snd (Map.toAscList bySequence))+  where+    add (ids, sequences) event@(LedgerEvent sequenceNumber identity _ _ _ _ _ _ _ _) = do+      validateEvent event+      case Map.lookup identity ids of Just old | old /= event -> Left EventIdentityConflict; _ -> Right ()+      case Map.lookup sequenceNumber sequences of Just old | old /= event -> Left SequenceConflict; _ -> Right ()+      pure (Map.insert identity event ids, Map.insert sequenceNumber event sequences)++appendProjection :: Projection -> LedgerEvent -> Either LedgerError Projection+appendProjection state event@(LedgerEvent sequenceNumber identity ledger _ attempt _ _ _ context body) = do+  validateEvent event+  let LedgerPosition expected previous _ = position state+  require LedgerIdentityMismatch (ledger == expected)+  require SequenceGap (previous < maxBound && sequenceNumber == previous + 1)+  require EventIdentityConflict (all ((/= identity) . eventId) (events state))+  case body of+    StartMigration _ -> pure ()+    AuthorizeRecovery _ -> pure ()+    _ -> case attempt of+      Nothing -> pure ()+      Just current -> require ImpossibleTransition (all (\old -> runId old == runId event) [old | old <- events state, attemptId old == Just current])+  expectedContext <- case body of+    StartMigration definition -> Just <$> contextFor definition Nothing+    AdoptEntry definition _ -> Just <$> contextFor definition Nothing+    Initialize _ -> pure Nothing+    StartAdoption _ -> pure Nothing+    CompleteAdoption _ -> pure Nothing+    _ -> do+      (definition, _, _, _) <- getActive state+      let index = case body of+            StartStep i _ -> Just i+            SucceedStep i _ -> Just i+            FailStep i _ _ -> Just i+            AcceptCompletion i _ -> Just i+            AuthorizeRecovery (Recovery action _ _ _ _) -> case action of RetryStep i -> Just i; AcceptStep i -> Just i; _ -> Nothing+            _ -> Nothing+      Just <$> contextFor definition index+  require DefinitionMismatch (context == expectedContext)+  next <- transition state attempt body+  pure next {position = LedgerPosition ledger sequenceNumber identity, events = events state ++ [event]}++transition :: Projection -> Maybe Text -> EventBody -> Either LedgerError Projection+transition state attempt body = case body of+  Initialize _ -> Left ImpossibleTransition+  StartMigration definition -> do+    require UnresolvedHistory (isNothing (active state) && isNothing (adoption state))+    require DefinitionMismatch (ordinal definition == length (completed state))+    current <- maybe (Left ImpossibleTransition) Right attempt+    require EventIdentityConflict (current `notElem` allAttempts state)+    require DefinitionMismatch (all (\(RecordedMigration old _ _) -> (component old, migration old) /= (component definition, migration definition)) (completed state))+    pure state {active = Just (ActiveMigration definition current (replicate (length (steps definition)) NeverStarted) Nothing)}+  StartStep index query -> withActive state attempt $ \definition current checkpoints authorization -> do+    require ImpossibleTransition (nextUnresolved checkpoints == Just index && at index checkpoints == Just NeverStarted && query `notElem` allQueries state)+    require ImpossibleTransition (case authorization of Nothing -> True; Just Resume -> True; Just (RetryStep wanted) -> wanted == index; _ -> False)+    pure (ActiveMigration definition current (replace index (Started query) checkpoints) Nothing)+  SucceedStep index query -> withActive state attempt $ \definition current checkpoints authorization -> do+    require ImpossibleTransition (at index checkpoints == Just (Started query) && isNothing authorization)+    pure (ActiveMigration definition current (replace index (Succeeded query) checkpoints) Nothing)+  FailStep index query failure -> withActive state attempt $ \definition current checkpoints authorization -> do+    require ImpossibleTransition (at index checkpoints == Just (Started query) && isNothing authorization)+    pure (ActiveMigration definition current (replace index (Failed query failure) checkpoints) Nothing)+  CompleteMigration -> do+    (definition, current, checkpoints, authorization) <- getActive state+    require ImpossibleTransition (attempt == Just current && all resolvedStep checkpoints && (isNothing authorization || authorization == Just Finalize))+    pure state {completed = completed state ++ [RecordedMigration definition Applied checkpoints], active = Nothing}+  AuthorizeRecovery (Recovery action expected previous next (QuiescenceEvidence target _)) -> do+    (definition, current, checkpoints, _) <- getActive state+    require StaleLedgerState (expected == position state && previous == current && attempt == Just next)+    require LedgerIdentityMismatch (target == targetOf state)+    require EventIdentityConflict (next `notElem` allAttempts state)+    let unfinished = nextUnresolved checkpoints+        uncertain step = case at step checkpoints of Just (Started _) -> True; Just (Failed _ _) -> True; _ -> False+    case action of+      Resume -> require ImpossibleTransition (not (all resolvedStep checkpoints) && all (\step -> resolvedStep step || step == NeverStarted) checkpoints)+      RetryStep step -> require ImpossibleTransition (unfinished == Just step && uncertain step)+      AcceptStep step -> require ImpossibleTransition (unfinished == Just step && uncertain step)+      Finalize -> require ImpossibleTransition (all resolvedStep checkpoints)+    let revised = case action of RetryStep step -> replace step NeverStarted checkpoints; _ -> checkpoints+    pure state {active = Just (ActiveMigration definition next revised (Just action))}+  AcceptCompletion index evidence -> withActive state attempt $ \definition current checkpoints authorization -> do+    require ImpossibleTransition (authorization == Just (AcceptStep index) && nextUnresolved checkpoints == Just index)+    pure (ActiveMigration definition current (replace index (Accepted evidence) checkpoints) Nothing)+  StartAdoption batch@(AdoptionBatch _ _ _ _ count) -> do+    require ImpossibleTransition (null (completed state) && isNothing (active state) && isNothing (adoption state) && count > 0 && isNothing attempt)+    -- Adoption is a fresh-ledger batch, not a second import after a completed batch.+    require ImpossibleTransition (all (\event -> case bodyOf event of StartAdoption _ -> False; _ -> True) (events state))+    pure state {adoption = Just (ActiveAdoption batch [])}+  AdoptEntry definition evidence -> case adoption state of+    Nothing -> Left ImpossibleTransition+    Just (ActiveAdoption batch@(AdoptionBatch _ _ _ _ count) entries) -> do+      require ImpossibleTransition (isNothing attempt && length entries < count && ordinal definition == length entries)+      require DefinitionMismatch (all (\(RecordedMigration old _ _) -> (component old, migration old) /= (component definition, migration definition)) entries)+      pure state {adoption = Just (ActiveAdoption batch (entries ++ [RecordedMigration definition Adopted (replicate (length (steps definition)) (Accepted evidence))]))}+  CompleteAdoption digest -> case adoption state of+    Nothing -> Left ImpossibleTransition+    Just (ActiveAdoption (AdoptionBatch _ _ mapping definitionsHash count) entries) -> do+      require ImpossibleTransition (isNothing attempt && count == length entries && digest == mapping && definitionsDigest [definition | RecordedMigration definition _ _ <- entries] == definitionsHash)+      pure state {completed = entries, adoption = Nothing}+  where+    targetOf (Projection _ target _ _ _ _) = target+    bodyOf (LedgerEvent _ _ _ _ _ _ _ _ _ value) = value++withActive :: Projection -> Maybe Text -> (Definition -> Text -> [StepState] -> Maybe RecoveryAction -> Either LedgerError ActiveMigration) -> Either LedgerError Projection+withActive state attempt action = do+  (definition, current, checkpoints, authorization) <- getActive state+  require ImpossibleTransition (attempt == Just current)+  changed <- action definition current checkpoints authorization+  pure state {active = Just changed}++getActive :: Projection -> Either LedgerError (Definition, Text, [StepState], Maybe RecoveryAction)+getActive state = case (active state, adoption state) of+  (Just (ActiveMigration definition attempt checkpoints authorization), Nothing) -> Right (definition, attempt, checkpoints, authorization)+  _ -> Left ImpossibleTransition++comparePlan :: MigrationPlan -> Projection -> Either LedgerError ()+comparePlan plan state = do+  let supplied = definitions plan+      recorded =+        [definition | RecordedMigration definition _ _ <- completed state]+          ++ [definition | Just (ActiveMigration definition _ _ _) <- [active state]]+          ++ concat [[definition | RecordedMigration definition _ _ <- entries] | Just (ActiveAdoption _ entries) <- [adoption state]]+  require DefinitionMismatch (recorded == take (length recorded) supplied)++unresolved :: Projection -> Bool+unresolved state = isJust (active state) || isJust (adoption state)++resolvedStep :: StepState -> Bool+resolvedStep (Succeeded _) = True+resolvedStep (Accepted _) = True+resolvedStep _ = False++nextUnresolved :: [StepState] -> Maybe Int+nextUnresolved = findIndex (not . resolvedStep)++at :: Int -> [a] -> Maybe a+at index values+  | index < 0 = Nothing+  | otherwise = case drop index values of value : _ -> Just value; _ -> Nothing++replace :: Int -> a -> [a] -> [a]+replace index value values = take index values ++ [value] ++ drop (index + 1) values++allAttempts :: Projection -> [Text]+allAttempts = concatMap (\event -> case body event of StartMigration _ -> maybe [] pure (attemptId event); AuthorizeRecovery (Recovery _ _ _ next _) -> [next]; _ -> []) . events++allQueries :: Projection -> [Text]+allQueries = concatMap (\event -> case body event of StartStep _ query -> [query]; _ -> []) . events++require :: LedgerError -> Bool -> Either LedgerError ()+require _ True = Right ()+require failure False = Left failure
+ src/ClickHouseMigrate/Internal/RepairTypes.hs view
@@ -0,0 +1,15 @@+module ClickHouseMigrate.Internal.RepairTypes where++import ClickHouseMigrate.Ledger (AppendReceipt, AttemptId, AuditContext, LedgerPosition, QuiescenceEvidence)+import ClickHouseMigrate.Report (StepAttempt)++data RecoveryEvidence = RecoveryEvidence !AuditContext !QuiescenceEvidence ![AppendReceipt]++data RepairRequest+  = ResumeRequest !LedgerPosition !AttemptId !RecoveryEvidence+  | RetryStepRequest !LedgerPosition !AttemptId !StepAttempt !RecoveryEvidence+  | AcceptStepRequest !LedgerPosition !AttemptId !StepAttempt !RecoveryEvidence+  | FinalizeRequest !LedgerPosition !AttemptId !RecoveryEvidence++data RepairRequestError = MissingRecoveryEvidence | InvalidStepAttempt | InvalidRecoveryTarget | TooManyPendingReceipts | ForeignLedgerReceipt+  deriving stock (Eq, Show)
+ src/ClickHouseMigrate/Internal/Reporting.hs view
@@ -0,0 +1,208 @@+-- | Convert validated storage into safe immutable observations, without SQL.+module ClickHouseMigrate.Internal.Reporting where++import ClickHouseMigrate.Internal.ExclusiveTypes (AccessDescription)+import ClickHouseMigrate.Internal.LedgerCodec (definitions)+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.TargetTypes qualified as T+import ClickHouseMigrate.Internal.Transport qualified as Transport+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Target (ServerIdentity, TargetKey)++summaryDefinition :: L.Definition -> R.DefinitionSummary+summaryDefinition (L.Definition ordinal componentIndex component dependencies migration checksum steps) =+  R.DefinitionSummary ordinal componentIndex component dependencies migration checksum [(name, digest) | L.StepDefinition name digest <- steps]++emptyHistory :: TargetKey -> R.InspectionAccess -> R.HistoryReport+emptyHistory target access = R.HistoryReport target Nothing R.LedgerUnknown Nothing access [] [] False++observationReport :: TargetKey -> Maybe ServerIdentity -> R.InspectionAccess -> Store.Observation -> R.HistoryReport+observationReport target server access = \case+  Store.Missing -> R.HistoryReport target server R.LedgerMissing Nothing access [] [] False+  Store.EmptyNamespace -> R.HistoryReport target server R.LedgerUninitialized Nothing access [] [] False+  Store.History projection -> fromProjection server access projection++fromProjection :: Maybe ServerIdentity -> R.InspectionAccess -> L.Projection -> R.HistoryReport+fromProjection server access projection =+  let L.Projection position@(L.LedgerPosition ledger _ _) target completed active adoption events = projection+      complete (L.RecordedMigration definition state checkpoints) =+        migrationSummary+          events+          definition+          (case state of L.Applied -> R.Applied; L.Adopted -> R.Adopted)+          (latestAttempt definition events)+          checkpoints+      running = case active of+        Nothing -> []+        Just (L.ActiveMigration definition attempt checkpoints _) ->+          [migrationSummary events definition (if any failed checkpoints then R.Failed else R.InProgress) (Just (L.AttemptId attempt)) checkpoints]+      adopting = case adoption of+        Nothing -> []+        Just (L.ActiveAdoption _ entries) -> [migrationSummary events definition R.InProgress Nothing checkpoints | L.RecordedMigration definition _ checkpoints <- entries]+   in R.HistoryReport+        target+        server+        (R.LedgerInitializedAt ledger)+        (Just position)+        access+        (map eventReport events)+        (map complete completed ++ running ++ adopting)+        (isJust active || isJust adoption)+  where+    failed (L.Failed _ _) = True; failed _ = False++latestAttempt :: L.Definition -> [L.LedgerEvent] -> Maybe L.AttemptId+latestAttempt definition events = case [L.AttemptId attempt | event <- events, sameMigration definition event, Just attempt <- [L.attemptId event]] of+  [] -> Nothing+  values -> Just (last values)++sameMigration :: L.Definition -> L.LedgerEvent -> Bool+sameMigration (L.Definition _ _ component _ migration _ _) event = case L.context event of+  Just (L.EventContext c m _ _) -> c == component && m == migration+  _ -> False++stepAttempt :: L.Definition -> Int -> Maybe Text -> [L.LedgerEvent] -> Maybe R.StepAttempt+stepAttempt definition index expected events = do+  event <- case [event | event <- reverse events, sameMigration definition event, L.StartStep i query <- [L.body event], i == index, maybe True (== query) expected] of+    value : _ -> Just value+    _ -> Nothing+  attempt <- L.AttemptId <$> L.attemptId event+  case L.body event of+    L.StartStep _ query -> case drop index (L.steps definition) of L.StepDefinition name _ : _ -> Just (R.StepAttempt attempt name (T.QueryId query)); _ -> Nothing+    _ -> Nothing++migrationSummary :: [L.LedgerEvent] -> L.Definition -> R.MigrationStatus -> Maybe L.AttemptId -> [L.StepState] -> R.MigrationSummary+migrationSummary events definition state attempt checkpoints =+  R.MigrationSummary+    (summaryDefinition definition)+    state+    attempt+    [ let (status, query, evidence) = case checkpoint of+            L.NeverStarted -> (R.StepPending, Nothing, Nothing)+            L.Started value -> (R.StepUncertain, Just value, Nothing)+            L.Succeeded value -> (R.StepObservedSuccess, Just value, Nothing)+            L.Failed value _ -> (R.StepObservedFailure, Just value, Nothing)+            L.Accepted (L.Evidence level references) -> (R.StepAccepted, Nothing, Just (R.EvidenceSummary level references))+       in R.StepSummary name checksum status (if status == R.StepPending then Nothing else stepAttempt definition index query events) evidence+    | (index, (L.StepDefinition name checksum, checkpoint)) <- zip [0 ..] (zip (L.steps definition) checkpoints)+    ]++pendingSummary :: L.Definition -> R.MigrationSummary+pendingSummary definition =+  R.MigrationSummary+    (summaryDefinition definition)+    R.Pending+    Nothing+    [R.StepSummary name checksum R.StepPending Nothing Nothing | L.StepDefinition name checksum <- L.steps definition]++inspectHistory :: Plan.MigrationPlan -> R.HistoryReport -> R.InspectionReport+inspectHistory plan history@(R.HistoryReport _ _ _ _ _ events recorded unresolved) =+  let supplied = definitions plan+      actual = [definition | R.MigrationSummary definition _ _ _ <- recorded]+      expected = map summaryDefinition supplied+      changes = [R.DefinitionChanged index (at index expected) (Just definition) | (index, definition) <- zip [0 ..] actual, at index expected /= Just definition]+      attempts = [R.UnresolvedAttempt attempt | R.MigrationSummary _ state (Just attempt) _ <- recorded, state `elem` [R.InProgress, R.Failed]]+      batches = if unresolved && null attempts then case reverse [batch | R.HistoryEvent _ _ _ _ _ _ _ _ _ (R.AdoptionBeginning batch) <- events] of batch : _ -> [R.IncompleteAdoption batch]; _ -> [] else []+      current =+        [ case at index recorded of+            Just value@(R.MigrationSummary observed _ _ _) | observed == definition -> value+            Just (R.MigrationSummary _ _ attempt steps) -> R.MigrationSummary definition R.Invalid attempt steps+            Nothing -> pendingSummary source+        | (index, (definition, source)) <- zip [0 ..] (zip expected supplied)+        ]+   in R.InspectionReport history current (changes ++ attempts ++ batches)++compatibilityError :: R.InspectionReport -> Maybe R.RunError+compatibilityError (R.InspectionReport _ _ issues)+  | any changed issues = Just (R.IncompatibleHistory issues)+  | not (null issues) = Just (R.LedgerFailed L.UnresolvedHistory)+  | otherwise = Nothing+  where+    changed (R.DefinitionChanged _ _ _) = True; changed _ = False++isCurrent :: R.InspectionReport -> Bool+isCurrent (R.InspectionReport _ migrations issues) = null issues && all (\(R.MigrationSummary _ state _ _) -> state `elem` [R.Applied, R.Adopted]) migrations++runHistory :: R.RunReport -> R.HistoryReport+runHistory (R.RunReport _ _ _ _ history _ _ _) = history++runPosition :: R.RunReport -> Maybe L.LedgerPosition+runPosition report = case runHistory report of R.HistoryReport _ _ _ position _ _ _ _ -> position++updateRun :: Plan.MigrationPlan -> R.HistoryReport -> R.RunReport -> R.RunReport+updateRun plan history (R.RunReport identity target access acquired _ _ observations initialized) = let R.InspectionReport _ migrations _ = inspectHistory plan history in R.RunReport identity target access acquired history migrations observations initialized++recordSql :: R.SqlObservation -> R.RunReport -> R.RunReport+recordSql value report@(R.RunReport _ _ _ _ _ _ observations _) = report {R.observations = observations ++ [value]}++initialRun :: Text -> TargetKey -> AccessDescription -> Plan.MigrationPlan -> R.RunReport+initialRun run target access plan =+  let history = emptyHistory target (R.GuardedObservation access)+      R.InspectionReport _ migrations _ = inspectHistory plan history+   in R.RunReport run target access False history migrations [] Nothing++historyPosition :: R.HistoryReport -> Maybe L.LedgerPosition+historyPosition (R.HistoryReport _ _ _ position _ _ _ _) = position++stepReference :: L.Definition -> Int -> Maybe R.StepReference+stepReference (L.Definition _ _ component _ migration checksum steps) index = case drop index steps of+  L.StepDefinition step digest : _ | index >= 0 -> Just (R.StepReference component migration step checksum digest)+  _ -> Nothing++eventReport :: L.LedgerEvent -> R.HistoryEvent+eventReport (L.LedgerEvent number event ledger run attempt time audit runner context body) =+  R.HistoryEvent+    number+    event+    ledger+    run+    (L.AttemptId <$> attempt)+    time+    audit+    runner+    (fmap (\(L.EventContext component migration checksum step) -> (component, migration, checksum, step)) context)+    (bodyReport body)++bodyReport :: L.EventBody -> R.HistoryBody+bodyReport = \case+  L.Initialize target -> R.LedgerInitialized target+  L.StartMigration definition -> R.MigrationDefinition (summaryDefinition definition)+  L.StartStep index query -> R.StepIntent index (T.QueryId query)+  L.SucceedStep index query -> R.StepCompletion index (T.QueryId query)+  L.FailStep index query (L.FailureEvidence code status server) -> R.ExecutionFailure index (T.QueryId query) (R.FailureSummary code status server)+  L.CompleteMigration -> R.MigrationCompletion+  L.AuthorizeRecovery (L.Recovery action position previous next (L.QuiescenceEvidence target audit)) -> R.RecoveryAuthorization (recoveryAction action) position (L.AttemptId previous) (L.AttemptId next) target audit+  L.AcceptCompletion index (L.Evidence level references) -> R.AcceptedCompletion index (R.EvidenceSummary level references)+  L.StartAdoption (L.AdoptionBatch source snapshot mapping definitionHash count) -> R.AdoptionBeginning (R.AdoptionSummary source snapshot mapping definitionHash count)+  L.AdoptEntry definition (L.Evidence level references) -> R.AdoptedDefinition (summaryDefinition definition) (R.EvidenceSummary level references)+  L.CompleteAdoption digest -> R.AdoptionCompletion digest++recoveryAction :: L.RecoveryAction -> R.RecoveryAction+recoveryAction = \case L.Resume -> R.Resume; L.RetryStep index -> R.RetryStep index; L.AcceptStep index -> R.AcceptStep index; L.Finalize -> R.Finalize++outcomeError :: Transport.TransportOutcome -> Maybe L.AppendReceipt -> R.RunError+outcomeError outcome receipt = case outcome of+  Transport.NotSubmitted failure -> R.LedgerRequestFailed Ledger.PositiveNonSubmission failure receipt+  Transport.ObservedServerFailure failure -> R.LedgerRequestFailed Ledger.ObservedFailure failure receipt+  Transport.SubmissionUncertain failure -> R.LedgerRequestFailed Ledger.UncertainSubmission failure receipt+  _ -> R.UnexpectedRuntimeFailure++storeError :: Store.StoreError -> R.RunError+storeError = \case+  Store.LedgerProblem failure -> R.LedgerFailed failure+  Store.AccessProblem failure -> R.AccessFailed failure+  Store.RequestProblem outcome receipt -> outcomeError outcome receipt+  Store.InitializationProblem ledger stage query outcome -> case outcomeError outcome Nothing of+    R.LedgerRequestFailed evidence failure _ -> R.InitializationFailed (ledgerIdText ledger) stage query evidence failure+    _ -> R.UnexpectedRuntimeFailure+  where+    ledgerIdText (T.LedgerId value) = value++at :: Int -> [a] -> Maybe a+at index values+  | index < 0 = Nothing+  | otherwise = case drop index values of value : _ -> Just value; _ -> Nothing
+ src/ClickHouseMigrate/Internal/RuntimeTypes.hs view
@@ -0,0 +1,15 @@+-- | Configuration only; no acquired capability is stored in a public context.+module ClickHouseMigrate.Internal.RuntimeTypes where++import ClickHouseMigrate.Event (DiagnosticPolicy, EventSink)+import ClickHouseMigrate.ExclusiveAccess (ExclusiveAccess)+import ClickHouseMigrate.Ledger (AuditContext, InitializationAuthorization)+import ClickHouseMigrate.Target (TargetConfig)++data MutationContext = MutationContext !TargetConfig !ExclusiveAccess !EventSink !DiagnosticPolicy++data UpOptions = UpOptions !AuditContext !(Maybe InitializationAuthorization)++data InspectionOptions = Observe | Guarded !ExclusiveAccess++data VerifyOptions = VerifyOptions !Bool deriving stock (Eq, Show)
+ src/ClickHouseMigrate/Internal/ServerPolicy.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE MultilineStrings #-}++-- | Pinned-server admission before application/ledger work. No initialization.+module ClickHouseMigrate.Internal.ServerPolicy+  ( TargetCheckError (..),+    probeTarget,+    withTransport,+    withTransportChecked,+    supportedVersions,+  )+where++import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.Http (Http, probeConnection, submit, withHttpAdmissionChecked)+import ClickHouseMigrate.Internal.TargetTypes+import ClickHouseMigrate.Internal.Transport+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (serverId)+import Data.Aeson (FromJSON, Key, Object, Result (..), fromJSON)+import Data.Aeson.KeyMap qualified as KM+import Data.IORef (atomicModifyIORef', newIORef)+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding++-- Exact tested-patch policy. Keep coordinated with the checked-in candidate matrix.+supportedVersions :: [Text]+supportedVersions = ["26.8.28.14"]++data TargetCheckError = ProbeFailure !TransportOutcome | TargetRejected !TransportError+  deriving stock (Eq, Show)++probeTarget :: Http -> TargetConfig -> QueryId -> IO (Either TargetCheckError ServerIdentity)+probeTarget http config identity = validateOutcome config <$> submit http identity JsonRows identityQuery++validateOutcome :: TargetConfig -> TransportOutcome -> Either TargetCheckError ServerIdentity+validateOutcome (TargetConfig _ _ expected _ _ _ _) outcome = case outcome of+  AcknowledgedSuccess (Rows [row]) -> validate row+  AcknowledgedSuccess _ -> rejected UnexpectedResponse+  failure -> Left (ProbeFailure failure)+  where+    rejected code = Left (TargetRejected (TransportError code Nothing Nothing))+    validate row+      | Set.fromList (KM.keys row) /= Set.fromList ["server_id", "version", "host", "async_insert", "mutations_sync", "alter_sync"] = rejected UnexpectedResponse+      | otherwise = case (get "server_id" row, get "version" row, get "host" row, get "async_insert" row, get "mutations_sync" row, get "alter_sync" row) of+          (Just uuid, Just version, Just host, Just (async :: Int), Just (mutation :: Int), Just (alter :: Int)) -> case serverId uuid of+            Left _ -> rejected UnexpectedResponse+            Right actual+              | actual /= expected -> rejected ServerIdentityMismatch+              | version `notElem` supportedVersions -> rejected UnsupportedServer+              | async /= 0 || mutation /= 1 || alter /= 1 -> rejected CompletionPolicyMismatch+              | otherwise -> Right (ServerIdentity actual version host)+          _ -> rejected UnexpectedResponse++get :: (FromJSON a) => Key -> Object -> Maybe a+get key row = KM.lookup key row >>= \value -> case fromJSON value of Success result -> Just result; Error _ -> Nothing++identityQuery :: ByteString+identityQuery =+  """+  SELECT toString(serverUUID()) AS server_id, version() AS version, hostName() AS host,+         toUInt8(getSetting('async_insert')) AS async_insert,+         toUInt8(getSetting('mutations_sync')) AS mutations_sync,+         toUInt8(getSetting('alter_sync')) AS alter_sync+  """++-- The only production constructor for a checked transport. Its Http resources+-- close at scope exit; a capability used after exit cannot submit again.+withTransport :: TargetConfig -> QueryId -> (Transport -> IO a) -> IO (ScopeResult (Either TargetCheckError a))+withTransport config probeId = withTransportChecked config probeId (pure (Right ()))++withTransportChecked :: TargetConfig -> QueryId -> IO (Either TransportError ()) -> (Transport -> IO a) -> IO (ScopeResult (Either TargetCheckError a))+withTransportChecked config probeId@(QueryId rootId) check action = do+  counter <- newIORef (0 :: Int)+  let prefix = "chm-admission-" <> Checksum.digestHex (Checksum.digest (Encoding.encodeUtf8 rootId)) <> "-"+      admit connection = do+        n <- atomicModifyIORef' counter (\old -> (old + 1, old))+        outcome <- probeConnection config connection (QueryId (prefix <> Text.pack (show n))) identityQuery+        pure $ case validateOutcome config outcome of+          Right _ -> Right ()+          Left (TargetRejected failure) -> Left failure+          Left (ProbeFailure (NotSubmitted failure)) -> Left failure+          Left (ProbeFailure (SubmissionUncertain failure)) -> Left failure+          Left (ProbeFailure (ObservedServerFailure failure)) -> Left failure+          Left _ -> Left (TransportError UnexpectedResponse Nothing Nothing)+  withHttpAdmissionChecked config admit check $ \http -> do+    checked <- probeTarget http config probeId+    case checked of+      Left failure -> pure (Left failure)+      Right identity -> Right <$> action (Transport (submit http) identity)
+ src/ClickHouseMigrate/Internal/SqlEnvelope.hs view
@@ -0,0 +1,212 @@+-- | Conservative lexical envelope; never a general SQL safety proof.+module ClickHouseMigrate.Internal.SqlEnvelope (validateSql, mentionsIdentifier) where++import ClickHouseMigrate.Error+import ClickHouseMigrate.Prelude+import Data.ByteString qualified as BS+import Data.Char (isAsciiLower, isAsciiUpper, toUpper)+import Data.List (find, isPrefixOf)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding++data Kind = Word !Text | Identifier !Text | Literal | Symbol !Char+  deriving stock (Eq, Show)++data Token = Token !Int !Kind deriving stock (Eq, Show)++failure :: DefinitionErrorCode -> Maybe Int -> Either DefinitionError a+failure c offset = Left (DefinitionError c ["sql"] offset)++validateSql :: ByteString -> Either DefinitionError ()+validateSql sql = do+  text <- either (const (failure InvalidUtf8 Nothing)) Right (Encoding.decodeUtf8' sql)+  if BS.elem 0 sql then failure NulInSql Nothing else pure ()+  tokens <- scan 0 (Text.unpack text)+  statement <- case break (\(Token _ k) -> k == Symbol ';') tokens of+    (xs, []) -> Right xs+    (xs, [_]) -> Right xs+    (_, Token n _ : _) -> failure MultipleStatements (Just n)+  case statement of+    [] -> failure EmptySql Nothing+    _ -> checkEnvelope statement++scan :: Int -> String -> Either DefinitionError [Token]+scan _ [] = Right []+scan n input@(c : cs)+  | c `elem` [' ', '\t', '\r', '\n', '\f'] = scan (n + 1) cs+  | "--" `isPrefixOf` input || "//" `isPrefixOf` input = lineComment+  | "#!" `isPrefixOf` input || "# " `isPrefixOf` input = lineComment+  | "/*" `isPrefixOf` input = do+      (used, rest) <- block n 1 2 (drop 2 input)+      scan (n + used) rest+  | c == '\'' || c == '"' || c == '`' = do+      (used, value, rest) <- quoted n c 1 [] cs+      let kind = if c == '\'' then Literal else Identifier (Text.pack (map toUpper value))+      (Token n kind :) <$> scan (n + used) rest+  | c == '$' = do+      let (tag, end) = span wordChar cs+      case end of+        '$' : rest | case tag of [] -> True; first : _ -> firstName first -> do+          let delimiter = '$' : tag ++ "$"+          (used, remaining) <- heredoc n delimiter (length delimiter) rest+          (Token n Literal :) <$> scan (n + used) remaining+        _ -> failure UnsupportedSql (Just n)+  | wordChar c =+      let (word, rest) = span wordChar input+       in (Token n (Word (Text.pack (map toUpper word))) :) <$> scan (n + length word) rest+  | c `elem` ("(),.;=+-*/%<>![]:{}?|&^~" :: String) = (Token n (Symbol c) :) <$> scan (n + 1) cs+  | otherwise = failure UnsupportedSql (Just n)+  where+    lineComment = let (comment, rest) = break (`elem` ['\r', '\n']) input in scan (n + length comment) rest++firstName :: Char -> Bool+firstName c = isAsciiLower c || isAsciiUpper c || c == '_'++wordChar :: Char -> Bool+wordChar c = firstName c || (c >= '0' && c <= '9')++block :: Int -> Int -> Int -> String -> Either DefinitionError (Int, String)+block start depth used input+  | "/*" `isPrefixOf` input = block start (depth + 1) (used + 2) (drop 2 input)+  | "*/" `isPrefixOf` input = if depth == 1 then Right (used + 2, drop 2 input) else block start (depth - 1) (used + 2) (drop 2 input)+block start _ _ [] = failure UnterminatedSql (Just start)+block start depth used (_ : rest) = block start depth (used + 1) rest++quoted :: Int -> Char -> Int -> String -> String -> Either DefinitionError (Int, String, String)+quoted start _ _ _ [] = failure UnterminatedSql (Just start)+quoted start quote used acc (c : rest)+  | c == quote = case rest of+      d : remaining | d == quote -> quoted start quote (used + 2) (quote : acc) remaining+      _ -> Right (used + 1, reverse acc, rest)+  | c == '\\' = case rest of+      [] -> failure UnterminatedSql (Just start)+      d : remaining+        | quote == '\'' || d `elem` [quote, '\\'] -> quoted start quote (used + 2) (d : acc) remaining+        -- Unfamiliar quoted-identifier escapes are outside the envelope.+        | otherwise -> failure UnsupportedSql (Just (start + used))+  | otherwise = quoted start quote (used + 1) (c : acc) rest++heredoc :: Int -> String -> Int -> String -> Either DefinitionError (Int, String)+heredoc _start delimiter used input+  | delimiter `isPrefixOf` input = Right (used + length delimiter, drop (length delimiter) input)+heredoc start _ _ [] = failure UnterminatedSql (Just start)+heredoc start delimiter used (_ : rest) = heredoc start delimiter (used + 1) rest++keyword :: Text -> Token -> Bool+keyword word (Token _ kind) = kind == Word word++isName :: Token -> Bool+isName (Token _ (Word w)) = maybe False (firstName . fst) (Text.uncons w)+isName (Token _ (Identifier _)) = True+isName _ = False++-- Objects in the command header must be qualified; deeper expressions remain+-- author obligations. Database DDL itself has a single database name.+qualifiedTarget :: [Token] -> Maybe [Token]+qualifiedTarget (a : Token _ (Symbol '.') : b : rest) | isName a && isName b = Just rest+qualifiedTarget _ = Nothing++skipIf :: [Token] -> [Token]+skipIf (a : b : c : rest) | and [keyword "IF" a, keyword "NOT" b, keyword "EXISTS" c] = rest+skipIf (a : b : rest) | keyword "IF" a && keyword "EXISTS" b = rest+skipIf rest = rest++checkEnvelope :: [Token] -> Either DefinitionError ()+checkEnvelope tokens = do+  case find protected tokens of+    Just (Token n _) -> failure ProtectedSetting (Just n)+    Nothing -> pure ()+  case find forbidden (localViewHeader tokens) of+    Just (Token n _) -> failure UnsupportedSql (Just n)+    Nothing -> pure ()+  if any onCluster (pairs tokens) || quotedSettings tokens then failure UnsupportedSql Nothing else pure ()+  if accepted (localViewHeader tokens) then Right () else failure UnsupportedSql (case tokens of Token n _ : _ -> Just n; _ -> Nothing)+  where+    protected (Token _ (Word w)) = w `elem` protectedNames+    protected (Token _ (Identifier w)) = w `elem` protectedNames+    protected _ = False+    forbidden (Token _ (Word w)) = w `elem` forbiddenWords || "REPLICATED" `Text.isPrefixOf` w+    forbidden (Token _ (Identifier w)) = w `elem` ["DISTRIBUTED", "REMOTE", "REMOTESECURE"] || "REPLICATED" `Text.isPrefixOf` w+    forbidden _ = False+    onCluster (a, b) = keyword "ON" a && keyword "CLUSTER" b+    quotedSettings xs = case dropWhile (not . keyword "SETTINGS") xs of+      [] -> False+      _ : rest -> any quotedAssignment (pairs rest)+    quotedAssignment (Token _ (Identifier _), Token _ (Symbol '=')) = True+    quotedAssignment (Token _ Literal, Token _ (Symbol '=')) = True+    quotedAssignment _ = False++-- Only this exact local view header permits REPLACE. ALTER replacement,+-- CREATE OR REPLACE TABLE and later REPLACE tokens remain excluded.+localViewHeader :: [Token] -> [Token]+localViewHeader (create : orToken : replace : view : rest)+  | and [keyword "CREATE" create, keyword "OR" orToken, keyword "REPLACE" replace, keyword "VIEW" view] = create : view : rest+localViewHeader tokens = tokens++protectedNames :: [Text]+protectedNames = ["ASYNC_INSERT", "MUTATIONS_SYNC", "ALTER_SYNC"]++forbiddenWords :: [Text]+forbiddenWords =+  [ "FORMAT",+    "INFILE",+    "OUTFILE",+    "ASYNC",+    "REFRESH",+    "WATCH",+    "REMOTE",+    "REMOTESECURE",+    "DISTRIBUTED",+    "ATTACH",+    "DETACH",+    "FETCH",+    "FREEZE",+    "UNFREEZE",+    "MOVE",+    "REPLACE",+    "LIVE",+    "WINDOW"+  ]++pairs :: [a] -> [(a, a)]+pairs xs = zip xs (drop 1 xs)++accepted :: [Token] -> Bool+accepted (a : b : rest)+  | keyword "CREATE" a = create (b : rest)+  | keyword "ALTER" a && keyword "TABLE" b = case qualifiedTarget (skipIf rest) of+      Just (action : _) -> any (`keyword` action) ["ADD", "DROP", "CLEAR", "MODIFY", "RENAME", "COMMENT", "MATERIALIZE", "UPDATE", "DELETE", "REMOVE"]+      _ -> False+  | keyword "DROP" a = object b (skipIf rest)+  | keyword "TRUNCATE" a && keyword "TABLE" b = maybe False (const True) (qualifiedTarget (skipIf rest))+  | keyword "EXCHANGE" a && keyword "TABLES" b = case qualifiedTarget rest of+      Just (andToken : second) | keyword "AND" andToken -> qualifiedTarget second == Just []+      _ -> False+  | keyword "RENAME" a && keyword "TABLE" b = maybe False (const True) (qualifiedTarget rest)+  | keyword "INSERT" a && keyword "INTO" b = case qualifiedTarget rest of+      Just body -> any (\t -> keyword "SELECT" t || keyword "VALUES" t) body+      _ -> False+  where+    create (kind : body) | keyword "MATERIALIZED" kind = case body of+      view : target | keyword "VIEW" view -> maybe False (const True) (qualifiedTarget (skipIf target))+      _ -> False+    create (kind : body) = object kind (skipIf body)+    create [] = False+    object kind body+      | keyword "TABLE" kind || keyword "VIEW" kind = maybe False (const True) (qualifiedTarget body)+      | keyword "DATABASE" kind = case body of name : _ -> isName name; _ -> False+      | otherwise = False+accepted _ = False++-- Runtime ledger namespace isolation reuses the SAME literal/comment-aware scan.+-- Quoted compound identifiers are conservatively split at dots. This is an+-- identifier reservation, not a proof about arbitrary deeper SQL expressions.+mentionsIdentifier :: Text -> ByteString -> Either DefinitionError Bool+mentionsIdentifier name sql = do+  text <- either (const (failure InvalidUtf8 Nothing)) Right (Encoding.decodeUtf8' sql)+  tokens <- scan 0 (Text.unpack text)+  let wanted = Text.toUpper name+      matches (Token _ (Word value)) = wanted == value+      matches (Token _ (Identifier value)) = wanted `elem` Text.splitOn "." value+      matches _ = False+  pure (any matches tokens)
+ src/ClickHouseMigrate/Internal/TargetTypes.hs view
@@ -0,0 +1,26 @@+-- | Private target representations. No caller-supplied Manager or Request.+module ClickHouseMigrate.Internal.TargetTypes where++import ClickHouseMigrate.Prelude++data TargetKey = TargetKey !Text deriving stock (Eq, Ord, Show)++data DirectEndpoint = DirectEndpoint !Text deriving stock (Eq, Show)++data ServerId = ServerId !Text deriving stock (Eq, Ord, Show)++data QueryId = QueryId !Text deriving stock (Eq, Ord, Show)++data LedgerId = LedgerId !Text deriving stock (Eq, Ord, Show)++data LedgerNamespace = LedgerNamespace !Text deriving stock (Eq, Ord, Show)++data LedgerBinding = FirstUse | ExistingLedger !LedgerId deriving stock (Eq, Show)++data Credentials = Credentials !ByteString !ByteString++data HttpLimits = HttpLimits !Int !Int deriving stock (Eq, Show)++data TargetConfig = TargetConfig !TargetKey !DirectEndpoint !ServerId !LedgerNamespace !LedgerBinding !Credentials !HttpLimits++data ServerIdentity = ServerIdentity !ServerId !Text !Text deriving stock (Eq, Show)
+ src/ClickHouseMigrate/Internal/Transport.hs view
@@ -0,0 +1,40 @@+-- | Four-way request evidence. No unsuccessful result authorizes replay.+module ClickHouseMigrate.Internal.Transport where++import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (QueryId, ServerIdentity)+import Data.Aeson (Object)++data RequestKind = Command | JsonRows | JsonRowStream deriving stock (Eq, Show)++data ResponsePayload = CommandAccepted | Rows ![Object] deriving stock (Eq, Show)++data TransportErrorCode = AdministrativeAccessLost | InvalidRequest | ClientScopeFailed | ConnectionFailed | DeadlineExpired | ResponseInterrupted | ResponseTooLarge | UnexpectedResponse | ServerRejected | RedirectRejected | ServerIdentityMismatch | UnsupportedServer | CompletionPolicyMismatch | CleanupFailed | CleanupTimedOut+  deriving stock (Eq, Ord, Show)++-- Only safe structured diagnostics; raw server/HTTP exceptions are never rendered.+data TransportError = TransportError+  { code :: !TransportErrorCode,+    httpStatus :: !(Maybe Int),+    serverCode :: !(Maybe Int)+  }+  deriving stock (Eq, Show)++data TransportOutcome+  = NotSubmitted !TransportError+  | AcknowledgedSuccess !ResponsePayload+  | ObservedServerFailure !TransportError+  | SubmissionUncertain !TransportError+  deriving stock (Eq, Show)++data ScopeResult a = ScopeResult !(Either TransportError a) ![TransportError]+  deriving stock (Eq, Show)++-- Private capability constructed only after target admission in ServerPolicy.+data Transport = Transport !(QueryId -> RequestKind -> ByteString -> IO TransportOutcome) !ServerIdentity++request :: Transport -> QueryId -> RequestKind -> ByteString -> IO TransportOutcome+request (Transport submit _) = submit++transportIdentity :: Transport -> ServerIdentity+transportIdentity (Transport _ identity) = identity
+ src/ClickHouseMigrate/Internal/Types.hs view
@@ -0,0 +1,33 @@+-- | Private representation. Public modules export types and read-only functions.+module ClickHouseMigrate.Internal.Types where++import ClickHouseMigrate.Prelude++-- Use data, not newtype: Coercible must not bypass validation.+data ComponentName = ComponentName !Text deriving stock (Eq, Ord, Show)++data MigrationName = MigrationName !Text deriving stock (Eq, Ord, Show)++data StepName = StepName !Text deriving stock (Eq, Ord, Show)++data MigrationId = MigrationId !ComponentName !MigrationName deriving stock (Eq, Ord, Show)++data SqlStep = SqlStep !StepName !ByteString deriving stock (Eq)++data Migration = Migration !MigrationName !(Maybe Text) !(NonEmpty SqlStep) deriving stock (Eq)++data MigrationComponent = MigrationComponent !ComponentName !(Set ComponentName) !(NonEmpty Migration) deriving stock (Eq)++data MigrationPlan = MigrationPlan !(NonEmpty ComponentView) !(NonEmpty MigrationView) deriving stock (Eq)++data ChecksumFormat = ChecksumV1 deriving stock (Eq, Ord, Show)++data StepChecksum = StepChecksum !ByteString deriving stock (Eq, Ord)++data MigrationChecksum = MigrationChecksum !ByteString deriving stock (Eq, Ord)++data ComponentView = ComponentView !ComponentName !(Set ComponentName) !(NonEmpty MigrationView) deriving stock (Eq)++data MigrationView = MigrationView !MigrationId !(Maybe Text) !(NonEmpty StepView) !MigrationChecksum deriving stock (Eq)++data StepView = StepView !StepName !ByteString !StepChecksum deriving stock (Eq)
+ src/ClickHouseMigrate/Ledger.hs view
@@ -0,0 +1,99 @@+-- | Validated administrative assertions and narrowly scoped ledger commands.+module ClickHouseMigrate.Ledger+  ( module ClickHouseMigrate.Internal.LedgerValidation,+    Reconciliation (PresentIdentical, Conflicting, NotObserved),+    initializeLedger,+    reconcileAppend,+    LedgerCommandResult (..),+    LedgerFailure (..),+    RequestEvidence (..),+    CleanupFailure (..),+    TransportError (..),+    TransportErrorCode (..),+  )+where++import ClickHouseMigrate.Internal.Exclusive (checkGuard, withExclusive)+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessDescription, AccessError (..), AccessResult (..), ExclusiveAccess (..), Guard)+import ClickHouseMigrate.Internal.Identity (freshUuid)+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.LedgerTypes+import ClickHouseMigrate.Internal.LedgerValidation+import ClickHouseMigrate.Internal.ServerPolicy (TargetCheckError (..), withTransportChecked)+import ClickHouseMigrate.Internal.TargetTypes (TargetConfig (..))+import ClickHouseMigrate.Internal.Transport (ScopeResult (..), Transport, TransportError (..), TransportErrorCode (..))+import ClickHouseMigrate.Internal.Transport qualified as Transport+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (QueryId, ledgerIdText, queryId)+import Control.Exception (SomeAsyncException, SomeException, fromException, throwIO, try)++-- These are safe structured observations. They carry no raw HTTP exceptions,+-- credentials, SQL, or server response bodies.+data RequestEvidence = PositiveNonSubmission | ObservedFailure | UncertainSubmission deriving stock (Eq, Show)++data LedgerFailure+  = LedgerRejected !LedgerError+  | AccessRejected !AccessError+  | RequestFailed !RequestEvidence !TransportError !(Maybe AppendReceipt)+  | InitializationFailed !Text !Text !QueryId !RequestEvidence !TransportError+  | AdministrativeScopeFailed+  deriving stock (Eq, Show)++data CleanupFailure = TransportCleanup !TransportError | AccessCleanup !AccessError deriving stock (Eq, Show)++data LedgerCommandResult a = LedgerCommandResult+  {result :: !(Either LedgerFailure a), access :: !AccessDescription, cleanupErrors :: ![CleanupFailure]}+  deriving stock (Eq, Show)++-- | Create/resume only this explicitly authorized target/UUID. On an initialized+-- matching ledger, return its observed position without appending or running SQL.+-- The caller must retain the authorization UUID after any interruption. Existing+-- binding mode never permits initialization of missing storage.+initializeLedger :: TargetConfig -> ExclusiveAccess -> InitializationAuthorization -> QuiescenceEvidence -> IO (LedgerCommandResult LedgerPosition)+initializeLedger config provider authorization quiet = command config provider $ \guard transport -> do+  run <- freshUuid+  store <- Store.openStore guard transport config (Just authorization) quiet run+  position <$> Store.storeProjection store++-- | Read-only exact-event reconciliation. NotObserved never authorizes a write.+reconcileAppend :: TargetConfig -> ExclusiveAccess -> AppendReceipt -> IO (LedgerCommandResult (Reconciliation, ExpectedLedgerState))+reconcileAppend config@(TargetConfig target _ _ _ _ _ _) provider receipt = command config provider $ \guard transport -> do+  (classification, observation) <- Store.reconcile guard transport config receipt+  pure (classification, case observation of Store.History state -> ExpectPosition (position state); _ -> ExpectAbsent target)++command :: TargetConfig -> ExclusiveAccess -> (Guard -> Transport -> IO a) -> IO (LedgerCommandResult a)+command config@(TargetConfig target _ _ _ _ _ _) provider@(ExclusiveAccess description _ _ _ _) action = do+  observed <- try @SomeException $ do+    probe <- freshUuid >>= either (const (fail "Invalid diagnostic identity")) pure . queryId . ("chm-" <>)+    withExclusive provider target $ \guard ->+      let check = checkGuard guard >>= pure . either (const (Left (TransportError AdministrativeAccessLost Nothing Nothing))) (const (Right ()))+       in withTransportChecked config probe check $ \transport -> Store.storeResult (action guard transport)+  case observed of+    Left exception | isJust (fromException exception :: Maybe SomeAsyncException) -> throwIO exception+    Left _ -> pure (LedgerCommandResult (Left AdministrativeScopeFailed) description [])+    Right (AccessResult outer accessCleanup) -> case outer of+      Left failure -> pure (LedgerCommandResult (Left (AccessRejected failure)) description (map AccessCleanup accessCleanup))+      Right (ScopeResult scoped transportCleanup) -> do+        let answer = case scoped of+              Left failure -> Left (RequestFailed UncertainSubmission failure Nothing)+              Right (Left (TargetRejected failure)) -> Left (RequestFailed PositiveNonSubmission failure Nothing)+              Right (Left (ProbeFailure outcome)) -> Left (outcomeFailure outcome Nothing)+              Right (Right (Left failure)) -> Left (storeFailure failure)+              Right (Right (Right value)) -> Right value+        pure (LedgerCommandResult answer description (map TransportCleanup transportCleanup ++ map AccessCleanup accessCleanup))++storeFailure :: Store.StoreError -> LedgerFailure+storeFailure = \case+  Store.LedgerProblem failure -> LedgerRejected failure+  Store.AccessProblem failure -> AccessRejected failure+  Store.RequestProblem outcome receipt -> outcomeFailure outcome receipt+  Store.InitializationProblem ledger stage identity outcome -> case outcomeFailure outcome Nothing of+    RequestFailed evidence failure _ -> InitializationFailed (ledgerIdText ledger) stage identity evidence failure+    _ -> AdministrativeScopeFailed++outcomeFailure :: Transport.TransportOutcome -> Maybe AppendReceipt -> LedgerFailure+outcomeFailure outcome receipt = case outcome of+  Transport.NotSubmitted failure -> RequestFailed PositiveNonSubmission failure receipt+  Transport.ObservedServerFailure failure -> RequestFailed ObservedFailure failure receipt+  Transport.SubmissionUncertain failure -> RequestFailed UncertainSubmission failure receipt+  Transport.AcknowledgedSuccess _ -> AdministrativeScopeFailed
+ src/ClickHouseMigrate/Plan.hs view
@@ -0,0 +1,195 @@+-- | Order-preserving validated plans and immutable context-bound views.+module ClickHouseMigrate.Plan+  ( MigrationPlan,+    migrationPlan,+    migrationPlanFromList,+    scheduledMigrationPlan,+    scheduledMigrationPlanFromList,+    inspectSchedule,+    inspectPlan,+    ComponentView,+    componentViewName,+    componentDependencies,+    componentMigrations,+    MigrationView,+    migrationId,+    migrationDescription,+    migrationSteps,+    migrationChecksum,+    StepView,+    stepViewName,+    stepSql,+    stepChecksum,+    StepChecksum,+    MigrationChecksum,+    ChecksumFormat (ChecksumV1),+    checksumFormatName,+    parseChecksumFormat,+    stepChecksumFormat,+    migrationChecksumFormat,+    stepChecksumHex,+    migrationChecksumHex,+    completionPolicy,+  )+where++import ClickHouseMigrate.Definition+import ClickHouseMigrate.Error+import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.Types+import ClickHouseMigrate.Prelude+import Control.Monad (foldM)+import Data.List.NonEmpty qualified as NE+import Data.Map.Strict qualified as Map+import Data.Set qualified as Set++migrationPlan :: NonEmpty MigrationComponent -> Either PlanError MigrationPlan+migrationPlan components = do+  let ordered = NE.toList components+      names = [name | MigrationComponent name _ _ <- ordered]+      allNames = Set.fromList names+      graph = Map.fromList [(name, deps) | MigrationComponent name deps _ <- ordered]+      bad c xs = Left (PlanError c (map componentNameText xs))+  if Set.size allNames /= length names then bad DuplicateComponent names else pure ()+  mapM_+    ( \(MigrationComponent name deps _) ->+        if Set.member name deps+          then bad SelfDependency [name]+          else case Set.toAscList (Set.difference deps allNames) of+            [] -> Right ()+            missing -> bad MissingDependency (name : missing)+    )+    ordered+  case cyclicRemainder graph of+    [] -> pure ()+    remaining -> bad DependencyCycle remaining+  let positions = Map.fromList (zip names [0 :: Int ..])+  mapM_+    ( \(MigrationComponent name deps _) ->+        case [dep | dep <- Set.toAscList deps, Map.lookup dep positions >= Map.lookup name positions] of+          [] -> Right ()+          late -> bad DependencyAfterDependent (name : late)+    )+    ordered+  let catalog = fmap bindComponent components+  pure (MigrationPlan catalog (catalog >>= componentMigrations))++-- | Parsing boundary for application formats which represent empty lists.+migrationPlanFromList :: [MigrationComponent] -> Either PlanError MigrationPlan+migrationPlanFromList = maybe (Left (PlanError EmptyPlan [])) migrationPlan . NE.nonEmpty++-- | Complete application-owned order. Dependencies constrain first activation;+-- later upgrades follow the explicit schedule, retaining each library's order.+scheduledMigrationPlan :: NonEmpty MigrationComponent -> NonEmpty MigrationId -> Either PlanError MigrationPlan+scheduledMigrationPlan components schedule = do+  ordinary <- migrationPlan components+  let catalog = inspectPlan ordinary+      authored = NE.toList (inspectSchedule ordinary)+      byId = Map.fromList [(migrationId view, view) | view <- authored]+      ids = NE.toList schedule+      named identity = [componentNameText (migrationIdComponent identity), migrationNameText (migrationIdName identity)]+      reject kind identity = Left (PlanError kind (named identity))+  resolved <- traverse (\identity -> maybe (reject UnknownScheduledMigration identity) Right (Map.lookup identity byId)) schedule+  _ <- foldM (\seen identity -> if Set.member identity seen then reject DuplicateScheduledMigration identity else Right (Set.insert identity seen)) Set.empty ids+  mapM_ (\view -> unless (migrationId view `elem` ids) (reject MissingScheduledMigration (migrationId view))) authored+  mapM_+    ( \component -> do+        let name = componentViewName component+            local = filter ((== name) . migrationIdComponent) ids+        unless (local == map migrationId (NE.toList (componentMigrations component))) (Left (PlanError MigrationOrderChanged [componentNameText name]))+    )+    catalog+  let dependencies = Map.fromList [(componentViewName c, componentDependencies c) | c <- NE.toList catalog]+  _ <-+    foldM+      ( \active identity -> do+          let name = migrationIdComponent identity+              missing = Map.findWithDefault Set.empty name dependencies `Set.difference` active+          unless (Set.member name active || Set.null missing) (Left (PlanError DependencyNotActivated (map componentNameText (name : Set.toAscList missing))))+          Right (Set.insert name active)+      )+      Set.empty+      ids+  pure (MigrationPlan catalog resolved)++scheduledMigrationPlanFromList :: [MigrationComponent] -> [MigrationId] -> Either PlanError MigrationPlan+scheduledMigrationPlanFromList components schedule = do+  catalog <- maybe (Left (PlanError EmptyPlan [])) Right (NE.nonEmpty components)+  order <- maybe (Left (PlanError EmptySchedule [])) Right (NE.nonEmpty schedule)+  scheduledMigrationPlan catalog order++inspectSchedule :: MigrationPlan -> NonEmpty MigrationView+inspectSchedule (MigrationPlan _ schedule) = schedule++cyclicRemainder :: Map.Map ComponentName (Set ComponentName) -> [ComponentName]+cyclicRemainder graph+  | Map.null graph = []+  | Map.null ready = Map.keys graph+  | otherwise = cyclicRemainder (Map.map (`Set.difference` Map.keysSet ready) waiting)+  where+    (ready, waiting) = Map.partition Set.null graph++bindComponent :: MigrationComponent -> ComponentView+bindComponent (MigrationComponent component deps migrations) = ComponentView component deps (fmap bindMigration migrations)+  where+    bindMigration (Migration name description steps) =+      let encodings = fmap (encodeStep name) steps+          boundSteps = NE.zipWith (\(SqlStep step sql) encoded -> StepView step sql (StepChecksum (Checksum.digest encoded))) steps encodings+          checksum = MigrationChecksum (Checksum.digest (Checksum.encodeMigrationV1 (componentNameText component) (migrationNameText name) (NE.toList encodings)))+       in MigrationView (MigrationId component name) description boundSteps checksum+    encodeStep migration (SqlStep step sql) = Checksum.encodeStepV1 (componentNameText component) (migrationNameText migration) (stepNameText step) sql++inspectPlan :: MigrationPlan -> NonEmpty ComponentView+inspectPlan (MigrationPlan components _) = components++componentViewName :: ComponentView -> ComponentName+componentViewName (ComponentView name _ _) = name++componentDependencies :: ComponentView -> Set ComponentName+componentDependencies (ComponentView _ deps _) = deps++componentMigrations :: ComponentView -> NonEmpty MigrationView+componentMigrations (ComponentView _ _ migrations) = migrations++migrationId :: MigrationView -> MigrationId+migrationId (MigrationView identity _ _ _) = identity++migrationDescription :: MigrationView -> Maybe Text+migrationDescription (MigrationView _ description _ _) = description++migrationSteps :: MigrationView -> NonEmpty StepView+migrationSteps (MigrationView _ _ steps _) = steps++migrationChecksum :: MigrationView -> MigrationChecksum+migrationChecksum (MigrationView _ _ _ checksum) = checksum++stepViewName :: StepView -> StepName+stepViewName (StepView name _ _) = name++stepSql :: StepView -> ByteString+stepSql (StepView _ sql _) = sql++stepChecksum :: StepView -> StepChecksum+stepChecksum (StepView _ _ checksum) = checksum++checksumFormatName :: ChecksumFormat -> Text+checksumFormatName ChecksumV1 = "sha256-v1"++parseChecksumFormat :: Text -> Either DefinitionError ChecksumFormat+parseChecksumFormat "sha256-v1" = Right ChecksumV1+parseChecksumFormat _ = Left (DefinitionError UnknownChecksumFormat ["checksum", "format"] Nothing)++stepChecksumFormat :: StepChecksum -> ChecksumFormat+stepChecksumFormat _ = ChecksumV1++migrationChecksumFormat :: MigrationChecksum -> ChecksumFormat+migrationChecksumFormat _ = ChecksumV1++stepChecksumHex :: StepChecksum -> Text+stepChecksumHex (StepChecksum bytes) = Checksum.digestHex bytes++migrationChecksumHex :: MigrationChecksum -> Text+migrationChecksumHex (MigrationChecksum bytes) = Checksum.digestHex bytes++completionPolicy :: ByteString+completionPolicy = Checksum.completionPolicyV1
+ src/ClickHouseMigrate/Prelude.hs view
@@ -0,0 +1,12 @@+{-# LANGUAGE PackageImports #-}++-- | Small project prelude. Extend this as shared dependencies are introduced.+module ClickHouseMigrate.Prelude (module X) where++import "base" Control.Monad as X (unless, void, when)+import "base" Data.List.NonEmpty as X (NonEmpty (..))+import "base" Data.Maybe as X (fromMaybe, isJust, isNothing)++import "bytestring" Data.ByteString as X (ByteString)+import "containers" Data.Set as X (Set)+import "text" Data.Text as X (Text)
+ src/ClickHouseMigrate/Repair.hs view
@@ -0,0 +1,202 @@+-- | Explicit audited recovery. No operation rewrites history or embedded SQL.+module ClickHouseMigrate.Repair+  ( RepairRequest,+    RecoveryEvidence,+    RepairRequestError (..),+    recoveryEvidence,+    resumeRequest,+    retryStepRequest,+    acceptStepRequest,+    finalizeRequest,+    repair,+  )+where++import ClickHouseMigrate.Definition qualified as Def+import ClickHouseMigrate.Event qualified as Event+import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.Execution+import ClickHouseMigrate.Internal.Identity (freshUuid)+import ClickHouseMigrate.Internal.LedgerCodec (encodeEvent)+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Preflight (validateRecoveryCapacity, validateRecoveryEvent)+import ClickHouseMigrate.Internal.Projection (resolvedStep)+import ClickHouseMigrate.Internal.RepairTypes+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Internal.RuntimeTypes (MutationContext (..))+import ClickHouseMigrate.Internal.TargetTypes qualified as T+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Target qualified as Target+import Data.List (findIndex)+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding++-- | External evidence is required even if no recorded query appears active.+-- Absence from system.processes is only a diagnostic; it cannot rule out a queued+-- or delayed request. Pending receipts, when available, are checked read-only.+recoveryEvidence :: Ledger.AuditContext -> Ledger.QuiescenceEvidence -> [Ledger.AppendReceipt] -> Either RepairRequestError RecoveryEvidence+recoveryEvidence audit quiet receipts+  | null (Ledger.auditReferences audit) = Left MissingRecoveryEvidence+  | length receipts > 32 = Left TooManyPendingReceipts+  | any (\(L.AppendReceipt target _ _ _ _ _) -> target /= Ledger.quiescenceTarget quiet) receipts = Left InvalidRecoveryTarget+  | otherwise = Right (RecoveryEvidence audit quiet receipts)++resumeRequest :: Ledger.LedgerPosition -> Ledger.AttemptId -> RecoveryEvidence -> Either RepairRequestError RepairRequest+resumeRequest position attempt evidence = validateReceipts position evidence >> Right (ResumeRequest position attempt evidence)++finalizeRequest :: Ledger.LedgerPosition -> Ledger.AttemptId -> RecoveryEvidence -> Either RepairRequestError RepairRequest+finalizeRequest position attempt evidence = validateReceipts position evidence >> Right (FinalizeRequest position attempt evidence)++retryStepRequest :: Ledger.LedgerPosition -> Ledger.AttemptId -> R.StepAttempt -> RecoveryEvidence -> Either RepairRequestError RepairRequest+retryStepRequest position attempt step evidence = validateReceipts position evidence >> validateStep step >> Right (RetryStepRequest position attempt step evidence)++acceptStepRequest :: Ledger.LedgerPosition -> Ledger.AttemptId -> R.StepAttempt -> RecoveryEvidence -> Either RepairRequestError RepairRequest+acceptStepRequest position attempt step evidence = validateReceipts position evidence >> validateStep step >> Right (AcceptStepRequest position attempt step evidence)++validateReceipts :: Ledger.LedgerPosition -> RecoveryEvidence -> Either RepairRequestError ()+validateReceipts (L.LedgerPosition ledger _ _) (RecoveryEvidence _ _ receipts)+  | any (\(L.AppendReceipt _ actual _ _ _ _) -> actual /= ledger) receipts = Left ForeignLedgerReceipt+  | otherwise = Right ()++validateStep :: R.StepAttempt -> Either RepairRequestError ()+validateStep (R.StepAttempt _ name _) = either (const (Left InvalidStepAttempt)) (const (Right ())) (Def.stepName name)++requestParts :: RepairRequest -> (Ledger.LedgerPosition, Ledger.AttemptId, RecoveryEvidence)+requestParts = \case+  ResumeRequest position attempt evidence -> (position, attempt, evidence)+  RetryStepRequest position attempt _ evidence -> (position, attempt, evidence)+  AcceptStepRequest position attempt _ evidence -> (position, attempt, evidence)+  FinalizeRequest position attempt evidence -> (position, attempt, evidence)++repair :: MutationContext -> Plan.MigrationPlan -> RepairRequest -> IO (Either R.RunFailure R.RunReport)+repair context@(MutationContext (T.TargetConfig target _ _ _ binding _ _) _ _ _) plan request =+  let (expected@(L.LedgerPosition ledger _ _), requested, RecoveryEvidence audit quiet receipts) = requestParts request+      prerequisite = case binding of+        T.FirstUse -> Left R.ExistingBindingRequired+        T.ExistingLedger actual | actual /= ledger -> Left (R.LedgerFailed L.LedgerIdentityMismatch)+        _ | Ledger.quiescenceTarget quiet /= target -> Left (R.RepairNotAllowed R.WrongQuiescenceTarget)+        _ -> Right ()+   in withRuntimeCheck prerequisite context audit Nothing (Just quiet) plan $ \runtime -> do+        projection <- runtimeProjection runtime+        unless (L.position projection == expected) (abort (R.StalePosition expected (Just (L.position projection))))+        (definition, current, checkpoints) <- case L.active projection of+          Just (L.ActiveMigration definition current checkpoints _) -> pure (definition, current, checkpoints)+          Nothing | isJust (L.adoption projection) -> abort (R.RepairNotAllowed R.AdoptionInProgress)+          _ -> abort (R.RepairNotAllowed R.NoActiveMigration)+        unless (Ledger.attemptIdText requested == current) (abort (R.AttemptMismatch requested (Just (L.AttemptId current))))+        mapM_ (checkReceipt projection) receipts+        let queries = Set.toAscList (Set.fromList ([T.QueryId query | event <- L.events projection, L.StartStep _ query <- [L.body event]] ++ [query | L.AppendReceipt _ _ _ _ _ query <- receipts]))+        active <- activeQueries runtime queries+        unless (null active) (abort (R.PriorQueriesActive active))+        migration <- case drop (L.ordinal definition) (orderedMigrations plan) of+          (actual, migration) : _ | actual == definition -> pure migration+          _ -> abort (R.LedgerFailed L.DefinitionMismatch)+        (action, selection, query) <- case request of+          ResumeRequest {} -> do+            unless (all (\step -> resolvedStep step || step == L.NeverStarted) checkpoints) (abort (R.RepairNotAllowed R.UncertainStepRemains))+            unless (any (== L.NeverStarted) checkpoints) (abort (R.RepairNotAllowed R.NoPendingSteps))+            pure (L.Resume, Nothing, Nothing)+          FinalizeRequest {} -> do+            unless (all resolvedStep checkpoints) (abort (R.RepairNotAllowed R.StepsNotResolved))+            pure (L.Finalize, Nothing, Nothing)+          RetryStepRequest _ _ step _ -> do+            (index, query) <- selectedStep projection definition checkpoints step+            pure (L.RetryStep index, Just index, Just query)+          AcceptStepRequest _ _ step _ -> do+            (index, query) <- selectedStep projection definition checkpoints step+            pure (L.AcceptStep index, Just index, Just query)+        next <- freshUuid+        bound <- contextAt definition selection+        let L.LedgerPosition _ number _ = L.position projection+            pending = [index | (index, checkpoint) <- zip [0 ..] checkpoints, checkpoint == L.NeverStarted]+            needed = case action of L.Resume -> 2 * toInteger (length pending) + 2; L.RetryStep _ -> 3; _ -> 2+        either abort pure (validateRecoveryCapacity number needed)+        authorization <- prepareEvent runtime projection (Just next) (L.AuthorizeRecovery (L.Recovery action expected current next quiet)) bound+        authorized <- preparedProjection R.RecoveryAuthorizationEvent projection authorization+        following <- case action of+          L.AcceptStep index -> do+            event <- prepareEvent runtime authorized (Just next) (L.AcceptCompletion index (L.Evidence L.OperatorAttestation (Ledger.auditReferences audit))) bound+            _ <- preparedProjection R.RecoveryAcceptanceEvent authorized event+            pure (Just event)+          L.Finalize -> do+            completion <- contextAt definition Nothing+            event <- prepareEvent runtime authorized (Just next) L.CompleteMigration completion+            _ <- preparedProjection R.RecoveryCompletionEvent authorized event+            pure (Just event)+          L.Resume -> checkExecutionMetadata runtime definition next pending True >> pure Nothing+          L.RetryStep index -> checkExecutionMetadata runtime definition next [index] False >> pure Nothing+        appendPreparedEvent runtime authorization+        emit runtime Event.RecoveryAuthorized (Just definition) selection query+        case (action, following) of+          (L.Resume, _) -> do+            mapM_ (\(index, (checkpoint, step)) -> when (checkpoint == L.NeverStarted) (executeStep runtime definition next index step)) (zip [0 ..] (zip checkpoints (NE.toList (Plan.migrationSteps migration))))+            finishMigration runtime definition next+          (L.RetryStep index, _) -> case drop index (NE.toList (Plan.migrationSteps migration)) of+            step : _ -> executeStep runtime definition next index step+            _ -> abort R.UnexpectedRuntimeFailure+          (L.AcceptStep index, Just event) -> do+            appendPreparedEvent runtime event+            emit runtime Event.StepAccepted (Just definition) (Just index) query+          (L.Finalize, Just event) -> do+            emit runtime Event.BeforeMigrationComplete (Just definition) Nothing Nothing+            appendPreparedEvent runtime event+            emit runtime Event.MigrationCompleted (Just definition) Nothing Nothing+          _ -> abort R.UnexpectedRuntimeFailure+  where+    checkReceipt projection (L.AppendReceipt receiptTarget ledger number identity digest _) = do+      let L.LedgerPosition actual _ _ = L.position projection+          matching = [event | event <- L.events projection, L.eventId event == identity || eventNumber event == number]+      unless (receiptTarget == target && ledger == actual) (abort (R.LedgerFailed L.LedgerIdentityMismatch))+      case matching of+        [] -> pure () -- Caller-established quiescence + fresh full guarded read.+        [event] | L.eventId event == identity && eventNumber event == number && Checksum.digestHex (Checksum.digest (encodeEvent event)) == digest -> pure ()+        _ -> abort (R.RepairNotAllowed R.ConflictingAppend)+    eventNumber (L.LedgerEvent number _ _ _ _ _ _ _ _ _) = number++-- SQL outcomes are not known yet. Bound every possible generated envelope using+-- maximal sequence digits, timestamp precision and safe numeric failure fields.+checkExecutionMetadata :: Runtime -> L.Definition -> Text -> [Int] -> Bool -> IO ()+checkExecutionMetadata runtime definition attempt indices completion = do+  projection <- runtimeProjection runtime+  let L.LedgerPosition ledger _ _ = L.position projection+      uuid = "00000000-0000-4000-8000-000000000000"+      query = "chm-" <> uuid+      check kind body context = either abort pure (validateRecoveryEvent kind (L.LedgerEvent maxBound uuid ledger (runtimeRunId runtime) (Just attempt) "2026-10-16T00:00:00.999999999999Z" (runtimeAudit runtime) "0.1.0.0" context body))+  mapM_+    ( \index -> do+        context <- contextAt definition (Just index)+        check R.RecoveryStepStartEvent (L.StartStep index query) context+        check R.RecoveryStepSuccessEvent (L.SucceedStep index query) context+        check R.RecoveryStepFailureEvent (L.FailStep index query (L.FailureEvidence "UnexpectedResponse" (Just 599) (Just maxBound))) context+    )+    indices+  when completion $ contextAt definition Nothing >>= check R.RecoveryCompletionEvent L.CompleteMigration++selectedStep :: L.Projection -> L.Definition -> [L.StepState] -> R.StepAttempt -> IO (Int, T.QueryId)+selectedStep projection definition checkpoints requested@(R.StepAttempt _ name query) = do+  index <- maybe (abort (R.StepAttemptMismatch requested)) pure (findIndex (\(L.StepDefinition actual _) -> actual == name) (L.steps definition))+  recorded <- case Reporting.at index checkpoints of+    Just (L.Started identity) -> pure identity+    Just (L.Failed identity _) -> pure identity+    _ -> abort (R.RepairNotAllowed R.SelectedStepIsResolved)+  unless (findIndex (not . resolvedStep) checkpoints == Just index && Target.queryIdText query == recorded && Reporting.stepAttempt definition index (Just recorded) (L.events projection) == Just requested) (abort (R.StepAttemptMismatch requested))+  pure (index, query)++activeQueries :: Runtime -> [T.QueryId] -> IO [T.QueryId]+activeQueries _ [] = pure []+activeQueries runtime queries = callStore $ do+  let sql = "SELECT query_id FROM system.processes WHERE query_id IN (" <> Text.intercalate "," ["'" <> Target.queryIdText query <> "'" | query <- queries] <> ")"+  rows <- Store.readRows (Just (runtimeGuard runtime)) (runtimeTransport runtime) (Encoding.encodeUtf8 sql)+  mapM+    ( \row -> do+        Store.keys ["query_id"] row+        value <- Store.member "query_id" row+        either (const (Store.abort (Store.LedgerProblem L.InvalidEvent))) pure (Target.queryId value)+    )+    rows
+ src/ClickHouseMigrate/Report.hs view
@@ -0,0 +1,204 @@+-- | Safe domain observations. Reports contain identities/checksums and audit+-- evidence, never SQL bodies, credentials, or raw server exception messages.+module ClickHouseMigrate.Report where++import ClickHouseMigrate.ExclusiveAccess (AccessDescription, AccessError)+import ClickHouseMigrate.Internal.AdoptionTypes (AdoptionError)+import ClickHouseMigrate.Ledger (AppendReceipt, AttemptId, AuditContext, EvidenceLevel, LedgerError, LedgerPosition, RequestEvidence, TransportError)+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Target (LedgerId, QueryId, ServerIdentity, TargetKey)+import Data.Word (Word64)++data StepReference = StepReference+  { component :: !Text,+    migration :: !Text,+    step :: !Text,+    migrationChecksum :: !Text,+    stepChecksum :: !Text+  }+  deriving stock (Eq, Show)++data DefinitionSummary = DefinitionSummary+  { ordinal :: !Int,+    componentIndex :: !Int,+    component :: !Text,+    dependencies :: ![Text],+    migration :: !Text,+    checksum :: !Text,+    stepDefinitions :: ![(Text, Text)]+  }+  deriving stock (Eq, Show)++data EvidenceSummary = EvidenceSummary !EvidenceLevel ![Text] deriving stock (Eq, Show)++data MigrationStatus = Pending | InProgress | Failed | Applied | Adopted | Invalid deriving stock (Eq, Show)++data StepStatus = StepPending | StepUncertain | StepObservedSuccess | StepObservedFailure | StepAccepted deriving stock (Eq, Show)++-- A step attempt is identified by its original migration attempt and query ID.+-- Recovery authorization may change the active migration attempt without sending+-- another query; reports preserve that distinction.+data StepAttempt = StepAttempt !AttemptId !Text !QueryId deriving stock (Eq, Show)++data StepSummary = StepSummary+  { name :: !Text,+    checksum :: !Text,+    state :: !StepStatus,+    attempt :: !(Maybe StepAttempt),+    evidence :: !(Maybe EvidenceSummary)+  }+  deriving stock (Eq, Show)++data MigrationSummary = MigrationSummary+  { definition :: !DefinitionSummary,+    state :: !MigrationStatus,+    attempt :: !(Maybe AttemptId),+    steps :: ![StepSummary]+  }+  deriving stock (Eq, Show)++data RecoveryAction = Resume | RetryStep !Int | AcceptStep !Int | Finalize deriving stock (Eq, Show)++data AdoptionSummary = AdoptionSummary+  { source :: !Text,+    snapshotDigest :: !Text,+    mappingDigest :: !Text,+    definitionsDigest :: !Text,+    count :: !Int+  }+  deriving stock (Eq, Show)++data FailureSummary = FailureSummary !Text !(Maybe Int) !(Maybe Int) deriving stock (Eq, Show)++data HistoryBody+  = LedgerInitialized !TargetKey+  | MigrationDefinition !DefinitionSummary+  | StepIntent !Int !QueryId+  | StepCompletion !Int !QueryId+  | ExecutionFailure !Int !QueryId !FailureSummary+  | MigrationCompletion+  | RecoveryAuthorization !RecoveryAction !LedgerPosition !AttemptId !AttemptId !TargetKey !AuditContext+  | AcceptedCompletion !Int !EvidenceSummary+  | AdoptionBeginning !AdoptionSummary+  | AdoptedDefinition !DefinitionSummary !EvidenceSummary+  | AdoptionCompletion !Text+  deriving stock (Eq, Show)++data HistoryEvent = HistoryEvent+  { sequence :: !Word64,+    eventId :: !Text,+    ledgerId :: !LedgerId,+    runId :: !Text,+    attempt :: !(Maybe AttemptId),+    occurredAt :: !Text,+    audit :: !AuditContext,+    runnerVersion :: !Text,+    context :: !(Maybe (Text, Text, Text, Maybe (Text, Text))),+    body :: !HistoryBody+  }+  deriving stock (Eq, Show)++data LedgerPresence = LedgerUnknown | LedgerMissing | LedgerUninitialized | LedgerInitializedAt !LedgerId deriving stock (Eq, Show)++data InspectionAccess = UnguardedObservation | GuardedObservation !AccessDescription deriving stock (Eq, Show)++data HistoryReport = HistoryReport+  { target :: !TargetKey,+    server :: !(Maybe ServerIdentity),+    presence :: !LedgerPresence,+    position :: !(Maybe LedgerPosition),+    access :: !InspectionAccess,+    events :: ![HistoryEvent],+    migrations :: ![MigrationSummary],+    unresolved :: !Bool+  }+  deriving stock (Eq, Show)++data CompatibilityIssue+  = DefinitionChanged !Int !(Maybe DefinitionSummary) !(Maybe DefinitionSummary)+  | UnresolvedAttempt !AttemptId+  | IncompleteAdoption !AdoptionSummary+  deriving stock (Eq, Show)++data InspectionReport = InspectionReport+  {history :: !HistoryReport, migrations :: ![MigrationSummary], issues :: ![CompatibilityIssue]}+  deriving stock (Eq, Show)++data VerificationReport = VerificationReport+  {inspection :: !InspectionReport, current :: !Bool}+  deriving stock (Eq, Show)++data SqlObservation = SqlObservation !StepReference !QueryId !SqlEvidence deriving stock (Eq, Show)++data SqlEvidence = QueryAcknowledged | QueryRejected !RequestEvidence !TransportError deriving stock (Eq, Show)++data RunReport = RunReport+  { runId :: !Text,+    target :: !TargetKey,+    access :: !AccessDescription,+    acquired :: !Bool,+    history :: !HistoryReport,+    migrations :: ![MigrationSummary],+    observations :: ![SqlObservation],+    initialized :: !(Maybe LedgerId)+  }+  deriving stock (Eq, Show)++data CallbackPhase = EventCallback !Text | SqlDiagnosticCallback deriving stock (Eq, Show)++data RepairReason = NoActiveMigration | AdoptionInProgress | UncertainStepRemains | NoPendingSteps | SelectedStepIsResolved | StepsNotResolved | ConflictingAppend | WrongQuiescenceTarget deriving stock (Eq, Show)++-- | Safe event roles, never user evidence or SQL text.+data RecoveryEventKind = RecoveryAuthorizationEvent | RecoveryAcceptanceEvent | RecoveryCompletionEvent | RecoveryStepStartEvent | RecoveryStepSuccessEvent | RecoveryStepFailureEvent+  deriving stock (Eq, Ord, Show)++data RunError+  = InvalidRuntimePlan !StepReference+  | LedgerNamespaceInSql !StepReference+  | DefinitionTooLarge !DefinitionSummary+  | AccessFailed !AccessError+  | TargetFailed !TransportError+  | LedgerFailed !LedgerError+  | LedgerRequestFailed !RequestEvidence !TransportError !(Maybe AppendReceipt)+  | InitializationFailed !Text !Text !QueryId !RequestEvidence !TransportError+  | SqlFailed !StepReference !QueryId !RequestEvidence !TransportError+  | CallbackFailed !CallbackPhase+  | IncompatibleHistory ![CompatibilityIssue]+  | VerificationNotCurrent+  | RecoveryEventTooLarge !RecoveryEventKind+  | AuditTooLarge+  | LedgerRequired+  | AdoptionRejected !AdoptionError+  | ExistingBindingRequired+  | StalePosition !LedgerPosition !(Maybe LedgerPosition)+  | AttemptMismatch !AttemptId !(Maybe AttemptId)+  | StepAttemptMismatch !StepAttempt+  | RepairNotAllowed !RepairReason+  | PriorQueriesActive ![QueryId]+  | CleanupFailed+  | UnexpectedRuntimeFailure+  deriving stock (Eq, Show)++data SecondaryError = RecordingError !RunError | ObserverError !CallbackPhase | TransportCleanup !TransportError | AccessCleanup !AccessError deriving stock (Eq, Show)++data RunFailure = RunFailure+  {primary :: !RunError, report :: !RunReport, secondary :: ![SecondaryError]}+  deriving stock (Eq, Show)++data InspectionFailure = InspectionFailure+  {primary :: !RunError, observation :: !(Maybe HistoryReport), secondary :: ![SecondaryError]}+  deriving stock (Eq, Show)++historyKind :: HistoryBody -> Text+historyKind = \case+  LedgerInitialized _ -> "initialized"+  MigrationDefinition _ -> "migration-started"+  StepIntent _ _ -> "step-started"+  StepCompletion _ _ -> "step-succeeded"+  ExecutionFailure _ _ _ -> "execution-failed"+  MigrationCompletion -> "migration-completed"+  RecoveryAuthorization {} -> "recovery-authorized"+  AcceptedCompletion _ _ -> "step-accepted"+  AdoptionBeginning _ -> "adoption-started"+  AdoptedDefinition _ _ -> "adoption-entry"+  AdoptionCompletion _ -> "adoption-completed"
+ src/ClickHouseMigrate/Runner.hs view
@@ -0,0 +1,36 @@+-- | Apply the ordered pending suffix once per authorized attempt. Ordinary up+-- never resumes an interrupted migration or adoption, nor retries an append.+module ClickHouseMigrate.Runner+  ( MutationContext,+    mutationContext,+    UpOptions,+    upOptions,+    up,+  )+where++import ClickHouseMigrate.Event (DiagnosticPolicy, EventSink)+import ClickHouseMigrate.ExclusiveAccess (ExclusiveAccess)+import ClickHouseMigrate.Internal.Execution+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Projection (unresolved)+import ClickHouseMigrate.Internal.RuntimeTypes+import ClickHouseMigrate.Ledger (AuditContext, InitializationAuthorization)+import ClickHouseMigrate.Plan (MigrationPlan)+import ClickHouseMigrate.Prelude+import ClickHouseMigrate.Report (RunError (LedgerFailed), RunFailure, RunReport)+import ClickHouseMigrate.Target (TargetConfig)++mutationContext :: TargetConfig -> ExclusiveAccess -> EventSink -> DiagnosticPolicy -> MutationContext+mutationContext = MutationContext++-- | Recurring runs should use Nothing. First use/resumption is a separate,+-- explicitly UUID-bound authorization, never a standing deployment default.+upOptions :: AuditContext -> Maybe InitializationAuthorization -> UpOptions+upOptions = UpOptions++up :: MutationContext -> UpOptions -> MigrationPlan -> IO (Either RunFailure RunReport)+up context (UpOptions audit authorization) plan = withRuntime context audit authorization Nothing plan $ \runtime -> do+  projection <- runtimeProjection runtime+  when (unresolved projection) (abort (LedgerFailed L.UnresolvedHistory))+  mapM_ (uncurry (executeMigration runtime)) (drop (length (L.completed projection)) (orderedMigrations plan))
+ src/ClickHouseMigrate/Target.hs view
@@ -0,0 +1,158 @@+-- | Explicit, validated configuration for one direct administrative target.+module ClickHouseMigrate.Target+  ( TargetKey,+    targetKey,+    targetKeyText,+    DirectEndpoint,+    directEndpoint,+    endpointText,+    ServerId,+    serverId,+    serverIdText,+    QueryId,+    queryId,+    queryIdText,+    LedgerId,+    ledgerId,+    newLedgerId,+    ledgerIdText,+    LedgerNamespace,+    ledgerNamespace,+    ledgerNamespaceText,+    LedgerBinding (FirstUse, ExistingLedger),+    Credentials,+    basicCredentials,+    HttpLimits,+    httpLimits,+    defaultHttpLimits,+    TargetConfig,+    targetConfig,+    configuredTargetKey,+    configuredLedgerBinding,+    TargetError (..),+    ServerIdentity,+    identityServerId,+    identityVersion,+    identityHostName,+  )+where++import ClickHouseMigrate.Internal.Identity (freshUuid)+import ClickHouseMigrate.Internal.TargetTypes+import ClickHouseMigrate.Prelude+import Data.Char (isAsciiLower, isAsciiUpper)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Text.Read (readMaybe)++data TargetError = InvalidTargetKey | InvalidEndpoint | InvalidServerId | InvalidQueryId | InvalidLedgerId | InvalidLedgerNamespace | InvalidCredentials | InvalidHttpLimits+  deriving stock (Eq, Ord, Show)++targetKey :: Text -> Either TargetError TargetKey+targetKey value+  | valid 256 (\c -> lowerDigit c || c `elem` ("/_:.-" :: String)) value && maybe False (lowerDigit . fst) (Text.uncons value) = Right (TargetKey value)+  | otherwise = Left InvalidTargetKey++directEndpoint :: Text -> Either TargetError DirectEndpoint+directEndpoint value = do+  authority <- case Text.stripPrefix "http://" value of+    Just rest -> Right rest+    Nothing -> maybe (Left InvalidEndpoint) Right (Text.stripPrefix "https://" value)+  let hostPort = maybe authority id (Text.stripSuffix "/" authority)+      validHost c = isAsciiLower c || isAsciiUpper c || digit c || c `elem` (".-" :: String)+      validPort "" = True+      validPort suffix = case Text.stripPrefix ":" suffix of+        Just port | not (Text.null port) && Text.all digit port -> case readMaybe (Text.unpack port) :: Maybe Int of+          Just n -> n > 0 && n <= 65535+          _ -> False+        _ -> False+      validAuthority = case Text.stripPrefix "[" hostPort of+        Just ipv6 -> case Text.breakOn "]" ipv6 of+          (host, suffix) -> not (Text.null host) && Text.all (\c -> digit c || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F') || c `elem` (":." :: String)) host && maybe False validPort (Text.stripPrefix "]" suffix)+        Nothing -> let (host, port) = Text.breakOn ":" hostPort in not (Text.null host) && Text.all validHost host && validPort port+  if Text.length value <= 2048 && validAuthority then Right (DirectEndpoint value) else Left InvalidEndpoint++serverId :: Text -> Either TargetError ServerId+serverId value = if uuid value then Right (ServerId value) else Left InvalidServerId++queryId :: Text -> Either TargetError QueryId+queryId value = if valid 128 (\c -> lowerDigit c || isAsciiUpper c || c `elem` ("_-" :: String)) value then Right (QueryId value) else Left InvalidQueryId++ledgerId :: Text -> Either TargetError LedgerId+ledgerId value = if uuid value then Right (LedgerId value) else Left InvalidLedgerId++ledgerNamespace :: Text -> Either TargetError LedgerNamespace+ledgerNamespace value+  | valid 128 (\c -> lowerDigit c || c == '_') value && maybe False (\(c, _) -> isAsciiLower c || c == '_') (Text.uncons value) = Right (LedgerNamespace value)+  | otherwise = Left InvalidLedgerNamespace++basicCredentials :: Text -> Text -> Either TargetError Credentials+basicCredentials user password+  | Text.null user || Text.any (`elem` (":\r\n\0" :: String)) user || Text.any (== '\0') password = Left InvalidCredentials+  | otherwise = Right (Credentials (Encoding.encodeUtf8 user) (Encoding.encodeUtf8 password))++-- | Deadline and raw-byte bound. Ordinary responses are bounded in total;+-- complete ledger reads apply this byte bound per JSON row (excluding LF).+-- Decoded history and reports still retain memory proportional to history size.+httpLimits :: Int -> Int -> Either TargetError HttpLimits+httpLimits timeoutMicros maxBytes+  | timeoutMicros > 0 && timeoutMicros <= 300000000 && maxBytes > 0 && maxBytes <= 16777216 = Right (HttpLimits timeoutMicros maxBytes)+  | otherwise = Left InvalidHttpLimits++defaultHttpLimits :: HttpLimits+defaultHttpLimits = HttpLimits 30000000 1048576++targetConfig :: TargetKey -> DirectEndpoint -> ServerId -> LedgerNamespace -> LedgerBinding -> Credentials -> HttpLimits -> TargetConfig+targetConfig = TargetConfig++targetKeyText :: TargetKey -> Text+targetKeyText (TargetKey value) = value++endpointText :: DirectEndpoint -> Text+endpointText (DirectEndpoint value) = value++serverIdText :: ServerId -> Text+serverIdText (ServerId value) = value++queryIdText :: QueryId -> Text+queryIdText (QueryId value) = value++ledgerIdText :: LedgerId -> Text+ledgerIdText (LedgerId value) = value++ledgerNamespaceText :: LedgerNamespace -> Text+ledgerNamespaceText (LedgerNamespace value) = value++valid :: Int -> (Char -> Bool) -> Text -> Bool+valid limit allowed value = not (Text.null value) && Text.length value <= limit && Text.all allowed value++lowerDigit :: Char -> Bool+lowerDigit c = isAsciiLower c || digit c++digit :: Char -> Bool+digit c = c >= '0' && c <= '9'++uuid :: Text -> Bool+uuid value = map Text.length (Text.splitOn "-" value) == [8, 4, 4, 4, 12] && Text.all (\c -> c == '-' || digit c || (c >= 'a' && c <= 'f')) value++identityServerId :: ServerIdentity -> ServerId+identityServerId (ServerIdentity value _ _) = value++identityVersion :: ServerIdentity -> Text+identityVersion (ServerIdentity _ value _) = value++identityHostName :: ServerIdentity -> Text+identityHostName (ServerIdentity _ _ value) = value++-- | Fresh client-generated UUID for a one-time first-use authorization. Persist+-- this value before submission and retain it after interrupted initialization.+newLedgerId :: IO LedgerId+newLedgerId = LedgerId <$> freshUuid++-- | Read-only nonsecret configuration identity for application adapters.+configuredTargetKey :: TargetConfig -> TargetKey+configuredTargetKey (TargetConfig key _ _ _ _ _ _) = key++-- | Binding mode is explicit; this accessor grants no initialization authority.+configuredLedgerBinding :: TargetConfig -> LedgerBinding+configuredLedgerBinding (TargetConfig _ _ _ _ binding _ _) = binding
+ test-driver/Main.hs view
@@ -0,0 +1,283 @@+-- | Opt-in qualification of actual private core sources. Never a public SQL API.+module Main (main) where++import ClickHouseMigrate.Definition qualified as Def+import ClickHouseMigrate.ExclusiveAccess+import ClickHouseMigrate.Internal.Exclusive+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessResult (..))+import ClickHouseMigrate.Internal.Http+import ClickHouseMigrate.Internal.Identity+import ClickHouseMigrate.Internal.LedgerCodec qualified as Codec+import ClickHouseMigrate.Internal.LedgerStore qualified as Store+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Projection qualified as Projection+import ClickHouseMigrate.Internal.ServerPolicy+import ClickHouseMigrate.Internal.TargetTypes qualified as TargetTypes+import ClickHouseMigrate.Internal.Transport+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Plan qualified as Plan+import ClickHouseMigrate.Target+import Control.Concurrent (forkIO, threadDelay)+import Control.Exception (AsyncException (ThreadKilled), SomeException, throwIO, try)+import Data.Aeson (Value (..), object, toJSON, (.=))+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString qualified as BS+import Data.ByteString.Lazy.Char8 qualified as Lazy+import Data.IORef (newIORef, readIORef, writeIORef)+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Maybe (fromMaybe)+import Data.Set qualified as Set+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import System.Environment (getArgs, lookupEnv)+import System.Exit (exitFailure)+import System.IO (hFlush, stdout)+import Text.Read (readMaybe)++main :: IO ()+main = do+  result <- try @SomeException run+  case result of+    Right () -> pure ()+    Left _ -> Lazy.putStrLn (Aeson.encode (object ["driverError" .= ("qualification-driver-failed" :: String)])) >> exitFailure++run :: IO ()+run = do+  args <- getArgs+  case args of+    [operation, endpoint, identity, file, microsText, bytesText] -> do+      micros <- maybe (fail "Invalid timeout") pure (readMaybe microsText)+      maxBytes <- maybe (fail "Invalid response bound") pure (readMaybe bytesText)+      user <- Text.pack . fromMaybe "default" <$> lookupEnv "CHM_TEST_USER"+      password <- Text.pack . fromMaybe "test-password" <$> lookupEnv "CHM_TEST_PASSWORD"+      expected <- Text.pack . fromMaybe "00000000-0000-0000-0000-000000000000" <$> lookupEnv "CHM_EXPECTED_SERVER_ID"+      namespace <- Text.pack . fromMaybe "migration_history" <$> lookupEnv "CHM_LEDGER_NAMESPACE"+      expectedLedger <- lookupEnv "CHM_EXPECTED_LEDGER_ID"+      binding <- maybe (pure FirstUse) (fmap ExistingLedger . either (const (fail "Invalid ledger binding")) pure . ledgerId . Text.pack) expectedLedger+      config <-+        either (const (fail "Invalid target")) pure $+          targetConfig+            <$> targetKey ("test-target/" <> namespace)+            <*> directEndpoint (Text.pack endpoint)+            <*> serverId expected+            <*> ledgerNamespace namespace+            <*> pure binding+            <*> basicCredentials user password+            <*> httpLimits micros maxBytes+      query <- either (const (fail "Invalid query id")) pure (queryId (Text.pack identity))+      sql <- BS.readFile file+      case operation of+        "ledger-smoke" -> ledgerSmoke config query+        mode | mode `elem` ["ledger-sequence", "ledger-recovery-sequence", "ledger-adoption-sequence", "ledger-read", "ledger-no-authorization"] -> ledgerProtocol mode config query+        mode | mode `elem` ["guarded-command", "guarded-reuse"] -> do+          probe <- either (const (fail "Invalid probe id")) pure (queryId (Text.pack (identity ++ "-identity")))+          result <- withTransport config probe $ \transport -> do+            first <- request transport query Command sql+            if mode == "guarded-command"+              then pure [first]+              else do+                secondId <- either (const (fail "Invalid second id")) pure (queryId (Text.pack (identity ++ "-second")))+                second <- request transport secondId Command sql+                pure [first, second]+          Lazy.putStrLn (Aeson.encode (guardedValue result))+        "cancel" -> do+          _ <- withHttp config $ \http -> do+            _ <- forkIO (submit http query Command sql >> pure ())+            threadDelay 250000+            throwIO ThreadKilled+          fail "Cancellation unexpectedly returned"+        "after-scope" -> do+          escaped <- withHttp config pure+          case escaped of+            ScopeResult (Right http) cleanup -> do+              outcome <- submit http query Command sql+              Lazy.putStrLn (Aeson.encode (scopeValue (ScopeResult (Right [outcome]) cleanup)))+            _ -> fail "Failed to create scope"+        _ -> do+          let scoped = case operation of+                "cleanup-failure" -> withHttpFinalizer config (fail "Synthetic cleanup failure")+                "cleanup-timeout" -> withHttpFinalizer config (threadDelay 2000000)+                _ -> withHttp config+          result <- scoped $ \http -> case operation of+            mode | mode `elem` ["command", "linger", "cleanup-failure", "cleanup-timeout"] -> (: []) <$> submit http query Command sql+            "read" -> (: []) <$> submit http query JsonRows sql+            "reuse" -> do+              first <- submit http query Command sql+              secondId <- either (const (fail "Invalid second id")) pure (queryId (Text.pack (identity ++ "-second")))+              second <- submit http secondId Command sql+              pure [first, second]+            _ -> fail "Unsupported driver operation"+          Lazy.putStrLn (Aeson.encode (scopeValue result))+          if operation == "linger" then hFlush stdout >> threadDelay 2000000 else pure ()+    _ -> fail "Expected operation endpoint query-id sql-file timeout-micros max-response-bytes"++scopeValue :: ScopeResult [TransportOutcome] -> Value+scopeValue (ScopeResult result cleanup) =+  object+    [ "outcomes" .= either (const ([] :: [Value])) (map outcomeValue) result,+      "scopeError" .= either (Just . errorValue) (const Nothing) result,+      "cleanupErrors" .= map errorValue cleanup+    ]++outcomeValue :: TransportOutcome -> Value+outcomeValue (NotSubmitted failure) = object ["outcome" .= ("not-submitted" :: String), "error" .= errorValue failure]+outcomeValue (SubmissionUncertain failure) = object ["outcome" .= ("submission-uncertain" :: String), "error" .= errorValue failure]+outcomeValue (ObservedServerFailure failure) = object ["outcome" .= ("observed-server-failure" :: String), "error" .= errorValue failure]+outcomeValue (AcknowledgedSuccess payload) = object ["outcome" .= ("acknowledged-success" :: String), "rows" .= case payload of CommandAccepted -> toJSON ([] :: [Value]); Rows rows -> toJSON rows]++errorValue :: TransportError -> Value+errorValue (TransportError code status serverCode) = object ["code" .= show code, "httpStatus" .= status, "serverCode" .= serverCode]++guardedValue :: ScopeResult (Either TargetCheckError [TransportOutcome]) -> Value+guardedValue (ScopeResult result cleanup) =+  object+    [ "outcomes" .= case result of Right (Right values) -> map outcomeValue values; _ -> [],+      "scopeError" .= case result of Left failure -> Just (errorValue failure); _ -> Nothing,+      "verificationError" .= case result of Right (Left failure) -> Just (checkValue failure); _ -> Nothing,+      "cleanupErrors" .= map errorValue cleanup+    ]++checkValue :: TargetCheckError -> Value+checkValue (TargetRejected failure) = object ["rejected" .= errorValue failure]+checkValue (ProbeFailure failure) = object ["probeFailure" .= outcomeValue failure]++-- The fixture owns a fresh isolated server, and this explicit provider represents+-- its single administrative process. It is not an application/default provider.+ledgerSmoke :: TargetConfig -> QueryId -> IO ()+ledgerSmoke config@(TargetTypes.TargetConfig key _ _ _ _ _ _) probe = do+  audit <- either (fail . show) pure (Ledger.auditContext "test operator's \"review\" 雪" "roundtrip\nencoding" ["owned isolated fixture"])+  quiet <- either (fail . show) pure (Ledger.attestQuiescence key audit)+  uuid <- freshUuid >>= either (fail . show) pure . ledgerId+  runIdentity <- freshUuid+  provider <- either (fail . show) pure (externallySerialized "owned-fixture-single-process" "complete isolated test target" 1000000 (const (pure ())) (const (pure ())) (const (pure True)))+  result <- withExclusive provider key $ \guard -> withTransport config probe $ \transport -> Store.storeResult $ do+    absent <- Store.load (Just guard) transport config+    unlessTest (absent == Store.Missing) "Fresh fixture already has a ledger"+    store <- Store.openStore guard transport config (Just (Ledger.authorizeInitialization key uuid audit)) quiet runIdentity+    initial <- Store.storeProjection store+    step <- either (fail . show) pure (Def.sqlStep "first" "CREATE TABLE app.events (id UInt8)")+    migration <- either (fail . show) pure (Def.sqlMigration "m" (step :| []))+    component <- either (fail . show) pure (Def.migrationComponent "app" Set.empty (migration :| []))+    plan <- either (fail . show) pure (Plan.migrationPlan (component :| []))+    definition <- case Codec.definitions plan of [value] -> pure value; _ -> fail "Unexpected test plan"+    identity <- freshUuid+    attempt <- freshUuid+    time <- diagnosticTime+    context <- either (fail . show) pure (Codec.contextFor definition Nothing)+    let event = L.LedgerEvent 2 identity uuid runIdentity (Just attempt) time audit "0.1.0.0" (Just context) (L.StartMigration definition)+    local <- Store.append store event+    loaded <- Store.load (Just guard) transport config+    unlessTest (loaded == Store.History local) "Stored event differs from projected event"+    pure (object ["position" .= Ledger.ledgerPositionText (L.position local), "initializedPosition" .= Ledger.ledgerPositionText (L.position initial), "roundtrip" .= True])+  case result of+    AccessResult (Right (ScopeResult (Right (Right (Right value))) [])) [] -> Lazy.putStrLn (Aeson.encode value)+    _ -> Lazy.putStrLn (Aeson.encode (object ["failure" .= show result])) >> exitFailure+  where+    unlessTest condition message = if condition then pure () else fail message++ledgerProtocol :: String -> TargetConfig -> QueryId -> IO ()+ledgerProtocol mode config@(TargetTypes.TargetConfig key _ _ namespace _ _ _) probe = do+  audit <- either (fail . show) pure (Ledger.auditContext "protocol test administrator" "one-shot ledger qualification" ["owned isolated fixture; predecessor requests are quiescent"])+  quiet <- either (fail . show) pure (Ledger.attestQuiescence key audit)+  requested <- lookupEnv "CHM_LEDGER_ID"+  identity <- maybe freshUuid (pure . Text.pack) requested >>= either (fail . show) pure . ledgerId+  runIdentity <- freshUuid+  attempt <- freshUuid+  provider <- either (fail . show) pure (externallySerialized "owned-fixture-single-administrator" "complete isolated test target" 1000000 (const (pure ())) (const (pure ())) (const (pure True)))+  poisoned <- newIORef False+  result <- withExclusive provider key $ \guard -> do+    let check = checkGuard guard >>= pure . either (const (Left (TransportError AdministrativeAccessLost Nothing Nothing))) (const (Right ()))+    withTransportChecked config probe check $ \transport ->+      Store.storeResult $+        if mode == "ledger-read"+          then observationValue <$> Store.load (Just guard) transport config+          else do+            let authorization = if mode == "ledger-no-authorization" then Nothing else Just (Ledger.authorizeInitialization key identity audit)+            store <- Store.openStore guard transport config authorization quiet runIdentity+            let sql = Encoding.encodeUtf8 ("CREATE TABLE default." <> ledgerNamespaceText namespace <> "_effects (id UInt8) ENGINE=MergeTree ORDER BY id")+            step <- either (fail . show) pure (Def.sqlStep "create" sql)+            migration <- either (fail . show) pure (Def.sqlMigration "one" (step :| []))+            component <- either (fail . show) pure (Def.migrationComponent "app" Set.empty (migration :| []))+            plan <- either (fail . show) pure (Plan.migrationPlan (component :| []))+            definition <- case Codec.definitions plan of [value] -> pure value; _ -> fail "Unexpected test definition"+            let append current body index = do+                  previous <- Store.storeProjection store+                  let L.LedgerPosition _ number _ = L.position previous+                  eventIdentity <- freshUuid+                  time <- diagnosticTime+                  context <- case body of L.StartAdoption _ -> pure Nothing; L.AdoptEntry _ _ -> Just <$> either (fail . show) pure (Codec.contextFor definition Nothing); L.CompleteAdoption _ -> pure Nothing; _ -> Just <$> either (fail . show) pure (Codec.contextFor definition index)+                  let event = L.LedgerEvent (number + 1) eventIdentity identity runIdentity current time audit "0.1.0.0" context body+                  submitted <- Store.storeResult (Store.append store event)+                  case submitted of+                    Right _ -> pure ()+                    Left failure@(Store.RequestProblem _ _) -> do+                      successor <- freshUuid+                      blocked <- Store.storeResult (Store.append store event {L.eventId = successor})+                      if blocked == Left failure then writeIORef poisoned True >> Store.abort failure else fail "Unacknowledged append did not poison store"+                    Left failure -> Store.abort failure+            if mode == "ledger-adoption-sequence"+              then do+                let digest = Text.replicate 64 "a"+                    evidence = L.Evidence L.OperatorAttestation ["protocol fixture: no predecessor checksum claim"]+                append Nothing (L.StartAdoption (L.AdoptionBatch "protocol-fixture" digest digest (Codec.definitionsDigest [definition]) 1)) Nothing+                append Nothing (L.AdoptEntry definition evidence) Nothing+                append Nothing (L.CompleteAdoption digest) Nothing+              else do+                append (Just attempt) (L.StartMigration definition) Nothing+                query <- Store.newQueryId+                append (Just attempt) (L.StartStep 0 (queryIdText query)) (Just 0)+                outcome <- request transport query Command (Plan.stepSql (NE.head (Plan.migrationSteps (NE.head (Plan.componentMigrations (NE.head (Plan.inspectPlan plan)))))))+                if mode == "ledger-recovery-sequence"+                  then do+                    failure <- case outcome of ObservedServerFailure value -> pure value; _ -> fail "Recovery fixture requires an observed server error"+                    append (Just attempt) (L.FailStep 0 (queryIdText query) (L.FailureEvidence (Text.pack (show (code failure))) (httpStatus failure) (serverCode failure))) (Just 0)+                    previous <- Store.storeProjection store+                    acceptedAttempt <- freshUuid+                    append (Just acceptedAttempt) (L.AuthorizeRecovery (L.Recovery (L.AcceptStep 0) (L.position previous) attempt acceptedAttempt quiet)) (Just 0)+                    append (Just acceptedAttempt) (L.AcceptCompletion 0 (L.Evidence L.OperatorAttestation ["fixture operator independently inspected preexisting intended table"])) (Just 0)+                    accepted <- Store.storeProjection store+                    finalAttempt <- freshUuid+                    append (Just finalAttempt) (L.AuthorizeRecovery (L.Recovery L.Finalize (L.position accepted) acceptedAttempt finalAttempt quiet)) Nothing+                    append (Just finalAttempt) L.CompleteMigration Nothing+                  else do+                    case outcome of AcknowledgedSuccess CommandAccepted -> pure (); _ -> Store.abort (Store.RequestProblem outcome Nothing)+                    append (Just attempt) (L.SucceedStep 0 (queryIdText query)) (Just 0)+                    append (Just attempt) L.CompleteMigration Nothing+            local <- Store.storeProjection store+            observed <- Store.load (Just guard) transport config+            if observed == Store.History local then pure (observationValue observed) else fail "Protocol read differs from local state"+  stopped <- readIORef poisoned+  let value = case ledgerProtocolValue result of Object fields -> Object (KM.insert "poisoned" (toJSON stopped) fields); other -> other+  Lazy.putStrLn (Aeson.encode value)++observationValue :: Store.Observation -> Value+observationValue = \case+  Store.Missing -> object ["observation" .= ("missing" :: String)]+  Store.EmptyNamespace -> object ["observation" .= ("uninitialized" :: String)]+  Store.History state -> let L.LedgerPosition ledger _ _ = L.position state in object ["observation" .= ("history" :: String), "position" .= Ledger.ledgerPositionText (L.position state), "ledgerId" .= ledgerIdText ledger, "eventCount" .= length (L.events state), "unresolved" .= Projection.unresolved state]++ledgerProtocolValue :: AccessResult (ScopeResult (Either TargetCheckError (Either Store.StoreError Value))) -> Value+ledgerProtocolValue (AccessResult acquired accessCleanup) = case acquired of+  Left failure -> object ["accessFailure" .= show failure, "accessCleanup" .= map show accessCleanup]+  Right (ScopeResult scoped cleanup) ->+    object+      [ "scopeError" .= either (Just . errorValue) (const Nothing) scoped,+        "verificationError" .= (case scoped of Right (Left failure) -> Just (checkValue failure); _ -> Nothing),+        "cleanupErrors" .= map errorValue cleanup,+        "accessCleanup" .= map show accessCleanup,+        "store"+          .= ( case scoped of+                 Right (Right (Right value)) -> object ["success" .= value]+                 Right (Right (Left failure)) -> storeErrorValue failure+                 _ -> object []+             )+      ]++storeErrorValue :: Store.StoreError -> Value+storeErrorValue = \case+  Store.LedgerProblem problem -> object ["ledgerError" .= show problem]+  Store.AccessProblem problem -> object ["accessError" .= show problem]+  Store.RequestProblem outcome receipt -> object ["request" .= outcomeValue outcome, "receipt" .= fmap Ledger.appendReceiptText receipt]+  Store.InitializationProblem ledger stage query outcome -> object ["initializationStage" .= stage, "ledgerId" .= ledgerIdText ledger, "queryId" .= queryIdText query, "request" .= outcomeValue outcome]
+ test/AdoptionTests.hs view
@@ -0,0 +1,89 @@+module AdoptionTests (adoptionTests) where++import ClickHouseMigrate+import ClickHouseMigrate.Adoption qualified as A+import ClickHouseMigrate.Internal.AdoptionTypes qualified as Internal+import ClickHouseMigrate.Internal.AdoptionValidation qualified as V+import ClickHouseMigrate.Internal.Checksum qualified as Hash+import ClickHouseMigrate.Internal.LedgerCodec qualified as Codec+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Reporting qualified as Reporting+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Report qualified as R+import Control.Monad (unless)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Set qualified as Set++assert :: String -> Bool -> IO ()+assert label condition = unless condition (fail ("adoption: " ++ label))++right :: (Show e) => Either e a -> IO a+right = either (fail . show) pure++adoptionTests :: IO ()+adoptionTests = do+  mapM_+    (\sql -> right (sqlStep "local-ddl" sql))+    ["CREATE OR REPLACE VIEW analytics.v AS SELECT 1", "EXCHANGE TABLES analytics.a AND analytics.b", "EXCHANGE TABLES `analytics`.`a` AND `analytics`.`b`"]+  mapM_+    (\sql -> assert "local DDL exception weakened exclusions" (case sqlStep "unsupported" sql of Left (DefinitionError UnsupportedSql _ _) -> True; _ -> False))+    ["CREATE OR REPLACE TABLE analytics.t (id UInt8) ENGINE=Memory", "CREATE OR REPLACE VIEW v AS SELECT 1", "ALTER TABLE analytics.t REPLACE PARTITION 1 FROM analytics.other", "EXCHANGE TABLES a AND analytics.b", "EXCHANGE TABLES analytics.a AND b", "EXCHANGE TABLES analytics.a AND analytics.b ON CLUSTER cluster", "EXCHANGE TABLES analytics.a AND analytics.b, analytics.c AND analytics.d", "SELECT throwIf(1)"]+  key <- right (targetKey "adoption/unit")+  other <- right (targetKey "adoption/foreign")+  ledger <- right (ledgerId "01234567-89ab-4def-8123-456789abcdef")+  audit <- right (Ledger.auditContext "unit operator" "reviewed fixture baseline" ["fixture:source"])+  quiet <- right (Ledger.attestQuiescence key audit)+  create <- right (sqlStep "create" "CREATE TABLE default.adoption (id UInt8) ENGINE=Memory")+  one <- right (sqlMigration "one" (create :| []))+  two <- right (sqlMigration "two" (create :| []))+  component <- right (migrationComponent "app" Set.empty (one :| [two]))+  plan <- right (migrationPlan (component :| []))+  (first, second) <- case Codec.definitions plan of+    [oneDefinition, twoDefinition] -> pure (oneDefinition, twoDefinition)+    _ -> fail "Unexpected adoption unit plan"+  let observation = R.HistoryReport key Nothing R.LedgerMissing Nothing R.UnguardedObservation [] [] False+      snapshot = Reporting.inspectHistory plan observation+      checksum definition = L.checksum definition+      build rows mappings = A.adoptionRequest key (Ledger.ExpectAbsent key) (Just (Ledger.authorizeInitialization key ledger audit)) quiet audit "legacy-fixture" "fixture:immutable-snapshot" rows mappings+      accept row migration digest = A.attestedMapping row "app" migration digest ["fixture:reviewed-step-mapping"]+  row <- right (A.legacyRow "001.sql" "2020-01-01T00:00:00Z" Nothing)+  later <- right (A.legacyRow "002.sql" "2020-01-02T00:00:00Z" Nothing)+  mapping <- right (accept "001.sql" "one" (checksum first))+  next <- right (accept "002.sql" "two" (checksum second))+  request <- right (build (row :| []) (mapping :| []))+  adopted <- right (A.planAdoption plan snapshot request)+  assert "independent adoption-v1 operator-attestation vector" (case A.inspectAdoption adopted of R.AdoptionSummary _ snapshotHash mappingHash definitionsHash 1 -> (snapshotHash, mappingHash, definitionsHash) == ("0cbd644a1b1871101f5944d249d1d48e6f83f487e43a427eced729390ba9e8f1", "12a26e9bb47128906ec2717968071b3b32ab931670ba83fbf4a8475b27394408", "efa03f51657e440641d3a2a303eebc9b0a933fac99f8ee1e4270e0d13416707c"); _ -> False)+  assert "filename-only history upgraded evidence" (case A.adoptionEntries adopted of [(_, R.EvidenceSummary Ledger.OperatorAttestation _)] -> True; _ -> False)+  let failure expected result = assert (show expected) (case result of Left actual -> actual == expected; Right _ -> False)+  right (build (row :| [row]) (mapping :| [])) >>= failure A.DuplicateSourceRow . A.planAdoption plan snapshot+  right (build (row :| []) (mapping :| [mapping])) >>= failure A.DuplicateMapping . A.planAdoption plan snapshot+  right (build (row :| [later]) (mapping :| [])) >>= failure A.IncompleteMapping . A.planAdoption plan snapshot+  unknown <- right (accept "unknown.sql" "one" (checksum first))+  right (build (row :| []) (unknown :| [])) >>= failure A.UnknownSourceRow . A.planAdoption plan snapshot+  right (build (row :| [later]) (next :| [mapping])) >>= failure A.MappingOrderMismatch . A.planAdoption plan snapshot+  changed <- right (accept "001.sql" "one" (checksum second))+  right (build (row :| []) (changed :| [])) >>= failure A.MappingDefinitionMismatch . A.planAdoption plan snapshot+  verified <- right (A.verifiedMapping "001.sql" "app" "one" (checksum first) "actual historical bytes" ["fixture:predecessor-checksum"])+  right (build (row :| []) (verified :| [])) >>= failure A.HistoricalChecksumRequired . A.planAdoption plan snapshot+  stored <- right (A.sourceChecksum "sha256" (Hash.digestHex (Hash.digest "actual historical bytes")))+  historical <- right (A.legacyRow "001.sql" "2020-01-01T00:00:00Z" (Just stored))+  good <- right (build (historical :| []) (verified :| [])) >>= right . A.planAdoption plan snapshot+  assert "independent adoption-v1 predecessor-checksum vector" (case A.inspectAdoption good of R.AdoptionSummary _ snapshotHash mappingHash definitionsHash 1 -> (snapshotHash, mappingHash, definitionsHash) == ("4906c7532c886734513805a2c730dcb0743e1e7a74ec621965a2b509985213bf", "32c3c3153d39088574f46249fa000b028e4c8353c9e04e6df7d49c054ecb5052", "efa03f51657e440641d3a2a303eebc9b0a933fac99f8ee1e4270e0d13416707c"); _ -> False)+  assert "actual predecessor checksum not distinguished" (case A.adoptionEntries good of [(_, R.EvidenceSummary Ledger.PredecessorChecksum _)] -> True; _ -> False)+  badBytes <- right (A.verifiedMapping "001.sql" "app" "one" (checksum first) "today's different bytes" ["fixture:predecessor-checksum"])+  right (build (historical :| []) (badBytes :| [])) >>= failure A.HistoricalChecksumMismatch . A.planAdoption plan snapshot+  failure A.UnsupportedSourceChecksum (A.sourceChecksum "unknown-algorithm" (checksum first))+  failure A.AdoptionTargetMismatch (A.adoptionRequest other (Ledger.ExpectAbsent other) (Just (Ledger.authorizeInitialization other ledger audit)) quiet audit "legacy-fixture" "fixture:snapshot" (row :| []) (mapping :| []))+  failure A.AdoptionInitializationRequired (A.adoptionRequest key (Ledger.ExpectAbsent key) Nothing quiet audit "legacy-fixture" "fixture:snapshot" (row :| []) (mapping :| []))+  -- A committed identical batch is read-only even with a stale absence token;+  -- a later native migration is retained without becoming a second import.+  let Internal.AdoptionPlan _ batch entries = adopted+      position = L.LedgerPosition ledger 5 "01234567-89ab-4def-8123-456789abcdee"+      L.AdoptionBatch source snapshotHash mappingHash definitionsHash count = batch+      event body = R.HistoryEvent 1 "01234567-89ab-4def-8123-456789abcdee" ledger "01234567-89ab-4def-8123-456789abcdee" Nothing "2026-10-16T00:00:00Z" audit "0.1.0.0" Nothing body+      recorded = [event (R.AdoptionBeginning (R.AdoptionSummary source snapshotHash mappingHash definitionsHash count))] ++ [event (R.AdoptedDefinition (Reporting.summaryDefinition definition) (R.EvidenceSummary level refs)) | (definition, L.Evidence level refs) <- entries] ++ [event (R.AdoptionCompletion mappingHash)]+      committed = R.HistoryReport key Nothing (R.LedgerInitializedAt ledger) (Just position) R.UnguardedObservation recorded [] False+  assert "identical committed batch not inspection-only" (V.checkHistory request batch entries committed == Right True)+  let changedRequest = A.adoptionRequest key (Ledger.ExpectAbsent key) (Just (Ledger.authorizeInitialization key ledger audit)) quiet audit "different-source" "fixture:immutable-snapshot" (row :| []) (mapping :| [])+  right changedRequest >>= failure A.AdoptionExpectedStateMismatch . A.planAdoption plan (Reporting.inspectHistory plan committed)+  putStrLn "Adoption mapping, prefix/order, source evidence strength and committed-batch identity checks passed."
+ test/ExclusiveTests.hs view
@@ -0,0 +1,52 @@+module ExclusiveTests (exclusiveTests) where++import ClickHouseMigrate.ExclusiveAccess+import ClickHouseMigrate.Internal.Exclusive+import ClickHouseMigrate.Internal.ExclusiveTypes (AccessResult (..))+import ClickHouseMigrate.Target (targetKey)+import Control.Concurrent (threadDelay)+import Control.Exception (AsyncException (ThreadKilled), SomeAsyncException, throwIO, try)+import Control.Monad (unless)+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)++assert :: String -> Bool -> IO ()+assert label condition = unless condition (fail ("exclusive: " ++ label))++right :: (Show e) => Either e a -> IO a+right = either (fail . show) pure++exclusiveTests :: IO ()+exclusiveTests = do+  target <- right (targetKey "unit/isolated")+  released <- newIORef (0 :: Int)+  let provider release = right (externallySerialized "controlled-unit-provider" "complete isolated target" 1000000 (const (pure ())) release (const (pure True)))+  normal <- provider (const (modifyIORef' released (+ 1)))+  result <- withExclusive normal target $ \guard -> checkGuard guard+  assert "normal scoped observation" (case result of AccessResult (Right (Right ())) [] -> True; _ -> False)+  readIORef released >>= assert "release once" . (== 1)+  escaped <- withExclusive normal target pure+  case escaped of+    AccessResult (Right guard) [] -> checkGuard guard >>= assert "escaped guard closed" . (== Left AccessScopeClosed)+    _ -> fail "Unexpected guard qualification result"+  cancelled <- try @SomeAsyncException (withExclusive normal target (const (throwIO ThreadKilled :: IO ())))+  assert "action cancellation rethrown" (either (const True) (const False) cancelled)+  readIORef released >>= assert "cancellation still releases" . (== 3)+  interruptedRelease <- provider (const (throwIO ThreadKilled))+  cleanupCancellation <- try @SomeAsyncException (withExclusive interruptedRelease target (const (fail "Primary synchronous failure" :: IO ())))+  assert "cleanup cancellation not swallowed by primary failure" (either (const True) (const False) cleanupCancellation)+  broken <- provider (const (fail "Synthetic release failure"))+  both <- withExclusive broken target (const (fail "Primary synchronous failure" :: IO ()))+  assert "primary and release failure separate" (case both of AccessResult (Left AccessActionFailed) [AccessReleaseFailed] -> True; _ -> False)+  delayed <- provider (const (threadDelay 2000000))+  bounded <- withExclusive delayed target (const (pure ()))+  assert "release deadline distinct from command success" (case bounded of AccessResult (Right ()) [AccessReleaseTimedOut] -> True; _ -> False)+  alive <- newIORef True+  lease <- right (externallySerialized "controlled-lease" "complete target" 1000000 (const (pure ())) (const (pure ())) (const (readIORef alive)))+  observed <- withExclusive lease target $ \guard -> do+    writeIORef alive False+    first <- checkGuard guard+    writeIORef alive True+    second <- checkGuard guard+    pure (first, second)+  assert "lost lease cannot resurrect" (case observed of AccessResult (Right (Left AccessLost, Left AccessLost)) [] -> True; _ -> False)+  putStrLn "Exclusive scope, cancellation, lease loss and separate bounded release checks passed."
+ test/LedgerTests.hs view
@@ -0,0 +1,167 @@+module LedgerTests (ledgerTests) where++import ClickHouseMigrate+import ClickHouseMigrate.Internal.LedgerCodec qualified as Codec+import ClickHouseMigrate.Internal.LedgerTypes qualified as L+import ClickHouseMigrate.Internal.Preflight (validateRecoveryCapacity, validateRecoveryEvent)+import ClickHouseMigrate.Internal.Projection qualified as Projection+import ClickHouseMigrate.Ledger qualified as Ledger+import ClickHouseMigrate.Report qualified as R+import Control.Monad (forM_, unless)+import Data.Aeson (object, (.=))+import Data.ByteString qualified as BS+import Data.List.NonEmpty (NonEmpty (..))+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Word (Word64)++assert :: String -> Bool -> IO ()+assert label predicate = unless predicate (fail ("ledger: " ++ label))++right :: (Show e) => Either e a -> IO a+right = either (fail . show) pure++ledgerTests :: IO ()+ledgerTests = do+  target <- right (targetKey "development/main")+  ledger <- right (ledgerId (uuid 99))+  audit <- right (Ledger.auditContext "operator's \"review\"" "test\nreason" ["mori://shinzui/clickhouse-migrate"])+  quiescence <- right (Ledger.attestQuiescence target audit)+  s1 <- right (sqlStep "first" "CREATE TABLE app.a (x UInt8)")+  s2 <- right (sqlStep "second" "ALTER TABLE app.a ADD COLUMN y UInt8")+  m <- right (sqlMigration "m" (s1 :| [s2]))+  c <- right (migrationComponent "app" Set.empty (m :| []))+  plan <- right (migrationPlan (c :| []))+  definition <- case Codec.definitions plan of [value] -> pure value; _ -> fail "Expected one test definition"+  recoveryEnvelopeTests target ledger definition+  let attempt = uuid 100+      event n a body = L.LedgerEvent n (uuid n) ledger (uuid 98) a "2026-10-16T00:00:00Z" audit "0.1.0.0" (eventContext body) body+      eventContext body = case body of+        L.Initialize _ -> Nothing+        L.StartAdoption _ -> Nothing+        L.CompleteAdoption _ -> Nothing+        _ ->+          either+            (error "Invalid test context")+            Just+            ( Codec.contextFor+                definition+                ( case body of+                    L.StartStep i _ -> Just i+                    L.SucceedStep i _ -> Just i+                    L.FailStep i _ _ -> Just i+                    L.AcceptCompletion i _ -> Just i+                    L.AuthorizeRecovery (L.Recovery action _ _ _ _) -> case action of L.RetryStep i -> Just i; L.AcceptStep i -> Just i; _ -> Nothing+                    _ -> Nothing+                )+            )+      initialized = event 1 Nothing (L.Initialize target)+      start = event 2 (Just attempt) (L.StartMigration definition)+      first = event 3 (Just attempt) (L.StartStep 0 "first-query")+      success = event 4 (Just attempt) (L.SucceedStep 0 "first-query")+      second = event 5 (Just attempt) (L.StartStep 1 "second-query")+      succeeded = event 6 (Just attempt) (L.SucceedStep 1 "second-query")+      complete = event 7 (Just attempt) L.CompleteMigration+      good = [initialized, start, first, success, second, succeeded, complete]+      fold = Projection.project target+      position n = L.LedgerPosition ledger n (uuid n)+      rejection label code result = assert label (either (== code) (const False) result)+  forM_ good $ \value -> assert "wire roundtrip" (Codec.decodeEvent (Codec.encodeEvent value) == Right value)+  golden <- BS.readFile "test/vectors/ledger-initialized-v1.json"+  assert "frozen independent JSON spelling" (Codec.encodeEvent initialized == BS.takeWhile (/= 10) golden)+  forM_ ["{}", Codec.encodeEvent initialized <> " ", "{\"version\":1,\"version\":1}", BS.take (BS.length (Codec.encodeEvent initialized) - 1) (Codec.encodeEvent initialized)] $ \bad ->+    assert "reject incomplete/noncanonical/duplicate fields" (either (const True) (const False) (Codec.decodeEvent bad))+  projection <- right (fold good)+  assert "applied and compatible" (not (Projection.unresolved projection) && Projection.comparePlan plan projection == Right ())+  assert "collapse only identical duplicates independent of storage order" (fold (reverse good ++ good) == Right projection)+  rejection "sequence conflict" L.SequenceConflict (fold (good ++ [first {L.eventId = uuid 77}]))+  rejection "identity conflict" L.EventIdentityConflict (fold (good ++ [first {L.sequence = 8}]))+  rejection "sequence gap" L.SequenceGap (fold [initialized, start, success])+  rejection "missing initialization" L.MissingInitialization (fold [start])+  rejection "unknown UUID binding" L.LedgerIdentityMismatch (fold [initialized, start {L.ledgerId = either (error "test UUID") id (ledgerId (uuid 88))}])+  rejection "success without intent" L.ImpossibleTransition (fold [initialized, start, success {L.sequence = 3, L.eventId = uuid 3}])+  rejection "out-of-order step" L.ImpossibleTransition (fold [initialized, start, first {L.body = L.StartStep 1 "second-query", L.context = eventContext (L.StartStep 1 "second-query")}])+  rejection "premature completion" L.ImpossibleTransition (fold [initialized, start, first {L.body = L.CompleteMigration, L.context = eventContext L.CompleteMigration}])+  forM_ [take 2 good, take 3 good, take 4 good, take 6 good] $ \prefix -> do+    state <- right (fold prefix)+    assert "every interrupted attempt blocks regardless of checkpoint" (Projection.unresolved state)+  changed <- right (migrationComponent "before" Set.empty (m :| []))+  inserted <- right (migrationPlan (changed :| [c]))+  rejection "component insertion shifts global prefix" L.DefinitionMismatch (Projection.comparePlan inserted projection)+  dep <- right (migrationComponent "app" (Set.singleton "before") (m :| []))+  changedDeps <- right (migrationPlan (changed :| [dep]))+  rejection "dependency/order mismatch" L.DefinitionMismatch (Projection.comparePlan changedDeps projection)+  let failed = event 4 (Just attempt) (L.FailStep 0 "first-query" (L.FailureEvidence "ServerRejected" (Just 500) (Just 60)))+  state <- right (fold [initialized, start, first, failed])+  assert "observed server failure unresolved" (Projection.unresolved state)+  let next = uuid 101+      authorize n previousPos action = event n (Just next) (L.AuthorizeRecovery (L.Recovery action previousPos attempt next quiescence))+      authorized = authorize 5 (position 4) (L.AcceptStep 0)+      accepted = event 6 (Just next) (L.AcceptCompletion 0 (L.Evidence L.OperatorAttestation ["operator inspected effects"]))+      resumed = event 7 (Just next) (L.StartStep 1 "after-accept")+      done = event 8 (Just next) (L.SucceedStep 1 "after-accept")+      final = event 9 (Just next) L.CompleteMigration+  recovered <- right (fold [initialized, start, first, failed, authorized, accepted, resumed, done, final])+  assert "accepted completion stays distinct" (case L.completed recovered of [L.RecordedMigration _ L.Applied (L.Accepted _ : _)] -> True; _ -> False)+  rejection "stale recovery position" L.StaleLedgerState (fold [initialized, start, first, failed, authorize 5 (position 3) (L.AcceptStep 0)])+  rejection "retry wrong step" L.ImpossibleTransition (fold [initialized, start, first, failed, authorize 5 (position 4) (L.RetryStep 1)])+  rejection "resume uncertain work" L.ImpossibleTransition (fold [initialized, start, first, failed, authorize 5 (position 4) L.Resume])+  retry <- right (fold [initialized, start, first, failed, authorize 5 (position 4) (L.RetryStep 0), event 6 (Just next) (L.StartStep 0 "new-query")])+  assert "retry uses a fresh query" (Projection.unresolved retry)+  rejection "query ID reuse" L.ImpossibleTransition (fold [initialized, start, first, failed, authorize 5 (position 4) (L.RetryStep 0), event 6 (Just next) (L.StartStep 0 "first-query")])+  let hash = Text.replicate 64 "a"+      batch = L.AdoptionBatch "legacy" hash hash (Codec.definitionsDigest [definition]) 1+      begin = event 2 Nothing (L.StartAdoption batch)+      entry = event 3 Nothing (L.AdoptEntry definition (L.Evidence L.OperatorAttestation ["historical filename only"]))+      end = event 4 Nothing (L.CompleteAdoption hash)+  partial <- right (fold [initialized, begin, entry])+  assert "incomplete adoption blocks" (Projection.unresolved partial)+  adopted <- right (fold [initialized, begin, entry, end])+  assert "complete adoption compatible" (not (Projection.unresolved adopted) && Projection.comparePlan plan adopted == Right ())+  rejection "incomplete batch cannot complete" L.ImpossibleTransition (fold [initialized, begin, end {L.sequence = 3, L.eventId = uuid 3}])+  rejection "adoption definition digest mismatch" L.ImpossibleTransition (fold [initialized, begin {L.body = L.StartAdoption (L.AdoptionBatch "legacy" hash hash hash 1)}, entry, end])+  request <- right (queryId "ledger-insert-query")+  let receipt = L.AppendReceipt target ledger 7 (uuid 7) (Text.replicate 64 "a") request+  assert "append receipt CLI roundtrip" (Ledger.parseAppendReceipt (Ledger.appendReceiptText receipt) == Right receipt && Ledger.appendQueryId receipt == request)+  forM_ [("\"version\":1", "\"version\":2"), ("\"kind\":\"initialized\"", "\"kind\":\"future-kind\"")] $ \(before, after) ->+    assert "unknown event version/kind blocks" (either (const True) (const False) (Codec.decodeEvent (Encoding.encodeUtf8 (Text.replace before after (Encoding.decodeUtf8 (Codec.encodeEvent initialized))))))+  assert "position CLI roundtrip" (Ledger.parseLedgerPosition (Ledger.ledgerPositionText (position 7)) == Right (position 7))+  assert "position text is canonical" (Ledger.parseLedgerPosition ("chm1:" <> ledgerIdText ledger <> ":007:" <> uuid 7) == Left L.InvalidPosition)+  assert "quiescence needs evidence" (either (== L.InvalidEvidence) (const False) (Ledger.auditContext "actor" "reason" [] >>= Ledger.attestQuiescence target))+  putStrLn "Ledger wire, corruption, global prefix, interrupted execution, recovery and adoption projection checks passed."++uuid :: Word64 -> Text+uuid n = "00000000-0000-4000-8000-" <> Text.justifyRight 12 '0' (Text.pack (show n))++recoveryEnvelopeTests :: TargetKey -> LedgerId -> L.Definition -> IO ()+recoveryEnvelopeTests target ledger definition = do+  assert "deterministic recovery capacity" (validateRecoveryCapacity (maxBound - 2) 2 == Right ())+  assert "capacity rejected before authorization" (validateRecoveryCapacity (maxBound - 1) 2 == Left (R.LedgerFailed L.LedgerLimitExceeded))+  let makeAudit actor reason count = right (Ledger.auditContext actor reason (replicate 15 (Text.replicate 2048 "x") ++ [Text.replicate count "x"]))+      context = either (error . show) id (Codec.contextFor definition (Just 0))+      envelope audit body = L.LedgerEvent 5 (uuid 5) ledger (uuid 90) (Just (uuid 91)) "2026-10-16T00:00:00.123456789012Z" audit "0.1.0.0" (Just context) body+      acceptance audit = envelope audit (L.AcceptCompletion 0 (L.Evidence L.OperatorAttestation (Ledger.auditReferences audit)))+  base <- makeAudit "op" "r" 1+  let remaining = Codec.maxEventBytes - BS.length (Codec.encodeEvent (acceptance base))+      actor = if odd remaining then "ops" else "op"+      count = 1 + remaining `div` 2+  exactAudit <- makeAudit actor "r" count+  tooLargeAudit <- makeAudit actor "rr" count+  let exact = acceptance exactAudit+      over = acceptance tooLargeAudit+  assert "exact canonical recovery byte bound" (BS.length (Codec.encodeEvent exact) == Codec.maxEventBytes && validateRecoveryEvent R.RecoveryAcceptanceEvent exact == Right ())+  assert "bound plus one typed error" (BS.length (Codec.encodeEvent over) == Codec.maxEventBytes + 1 && validateRecoveryEvent R.RecoveryAcceptanceEvent over == Left (R.RecoveryEventTooLarge R.RecoveryAcceptanceEvent))+  assert "canonical size roundtrip" (Codec.decodeEvent (Codec.encodeEvent exact) == Right exact)+  -- Even a worst possible JSON escape expansion of every event byte fits the+  -- default row limit. Valid canonical event bytes are a subset of this bound.+  let worstRow = Codec.encodeValue (object ["sequence" .= ("18446744073709551615" :: Text), "event_id" .= uuid 5, "ledger_id" .= uuid 99, "event" .= Text.replicate Codec.maxEventBytes "\SOH"])+  assert "default accommodates worst escaped row" (BS.length worstRow < 1048576)+  forM_ ["雪", "\"", "\\", "x\n"] $ \unit -> do+    audit <- right (Ledger.auditContext "op" "r" (replicate 16 (Text.replicate 1024 unit)))+    assert "UTF-8/JSON escaping counted" (validateRecoveryEvent R.RecoveryAcceptanceEvent (acceptance audit) == Left (R.RecoveryEventTooLarge R.RecoveryAcceptanceEvent))+  outer <- right (Ledger.auditContext "op" "r" (replicate 16 (Text.replicate 2048 "x")))+  quiet <- right (Ledger.attestQuiescence target outer)+  let authorization = envelope outer (L.AuthorizeRecovery (L.Recovery (L.AcceptStep 0) (L.LedgerPosition ledger 4 (uuid 4)) (uuid 92) (uuid 91) quiet))+  assert "combined independently valid audits" (validateRecoveryEvent R.RecoveryAuthorizationEvent authorization == Left (R.RecoveryEventTooLarge R.RecoveryAuthorizationEvent))
+ test/Main.hs view
@@ -0,0 +1,205 @@+module Main (main) where++import AdoptionTests (adoptionTests)+import ClickHouseMigrate+import ClickHouseMigrate.Internal.Checksum qualified as Checksum+import ClickHouseMigrate.Internal.Http (readJsonRowStream)+import ClickHouseMigrate.Internal.LedgerCodec qualified as Codec+import ClickHouseMigrate.Internal.Transport qualified as Transport+import Control.Monad (forM_, unless)+import Data.Aeson qualified as Aeson+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.Char (chr)+import Data.IORef (atomicModifyIORef', newIORef)+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.Encoding qualified as Encoding+import Data.Text.IO qualified as Text+import ExclusiveTests (exclusiveTests)+import LedgerTests (ledgerTests)+import Numeric (readHex)++assert :: String -> Bool -> IO ()+assert label condition = unless condition (fail label)++right :: (Show e) => Either e a -> IO a+right = either (fail . show) pure++isDefinitionError :: DefinitionErrorCode -> Either DefinitionError a -> Bool+isDefinitionError expected (Left (DefinitionError actual _ _)) = expected == actual+isDefinitionError _ _ = False++isPlanError :: PlanErrorCode -> Either PlanError a -> Bool+isPlanError expected (Left (PlanError actual _)) = expected == actual+isPlanError _ _ = False++main :: IO ()+main = do+  adoptionTests+  ledgerTests+  exclusiveTests+  streamValidation+  scheduleValidation+  targetValidation+  let sql = "CREATE TABLE app.events (id UInt64) ENGINE=MergeTree ORDER BY id"+  create <- right (sqlStep "create" sql)+  migration <- right (sqlMigration "0001" (create :| []))+  a <- right (migrationComponent "a" Set.empty (migration :| []))+  b <- right (migrationComponent "b" (Set.singleton "a") (migration :| []))+  plan <- right (migrationPlan (a :| [b]))+  assert "caller order preserved" (map (componentNameText . componentViewName) (NE.toList (inspectPlan plan)) == ["a", "b"])+  assert "component-bound checksums differ" (map (migrationChecksumHex . migrationChecksum . NE.head . componentMigrations) (NE.toList (inspectPlan plan)) /= replicate 2 (migrationChecksumHex (migrationChecksum (NE.head (componentMigrations (NE.head (inspectPlan plan)))))))+  assert "duplicate components" (isPlanError DuplicateComponent (migrationPlan (a :| [a])))+  assert "late dependency" (isPlanError DependencyAfterDependent (migrationPlan (b :| [a])))+  assert "missing dependency" (isPlanError MissingDependency (migrationPlan (b :| [])))+  self <- right (migrationComponent "a" (Set.singleton "a") (migration :| []))+  assert "self dependency" (isPlanError SelfDependency (migrationPlan (self :| [])))+  cycleA <- right (migrationComponent "a" (Set.singleton "b") (migration :| []))+  assert "cycle" (isPlanError DependencyCycle (migrationPlan (cycleA :| [b])))+  assert "duplicate steps" (isDefinitionError DuplicateName (sqlMigration "0001" (create :| [create])))+  assert "duplicate migrations" (isDefinitionError DuplicateName (migrationComponent "a" Set.empty (migration :| [migration])))+  assert "empty steps" (isDefinitionError EmptyCollection (sqlMigrationFromList "0001" []))+  assert "empty migrations" (isDefinitionError EmptyCollection (migrationComponentFromList "a" Set.empty []))+  assert "empty components" (isPlanError EmptyPlan (migrationPlanFromList []))+  forM_ [0 .. 255] $ \n -> do+    let c = chr n+        valid = (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9')+    assert ("single-character name " ++ show n) (either (const False) (const True) (stepName (Text.singleton c)) == valid)+  forM_ ["", "A", "a.b", "a/b", "_a", "-a", "雪", "a b", Text.replicate 129 "a"] $ \name ->+    assert "invalid name" (isDefinitionError InvalidName (sqlStep name sql))+  _ <- right (sqlStep (Text.replicate 128 "a") sql)+  forM_ ["", " \n\t", "-- comment", "/* outer /* inner */ end */", "# comment\n// comment", ";"] $ \empty ->+    assert "empty/comment-only SQL" (isDefinitionError EmptySql (sqlStep "x" empty))+  assert "invalid UTF-8" (isDefinitionError InvalidUtf8 (sqlStep "x" (BS.pack [255])))+  assert "NUL" (isDefinitionError NulInSql (sqlStep "x" (sql <> BS.singleton 0)))+  forM_ ["CREATE TABLE app.t (x UInt8); DROP TABLE app.t", "CREATE TABLE app.t (x UInt8);;", "INSERT INTO app.t VALUES ('a'); INSERT INTO app.t VALUES ('b')"] $ \multiple ->+    assert "multiple SQL statements" (isDefinitionError MultipleStatements (sqlStep "x" multiple))+  forM_ ["INSERT INTO app.t VALUES ('open", "INSERT INTO app.t VALUES ($x$open)", "CREATE TABLE app.t (x UInt8) /*", "CREATE TABLE app.\"t (x UInt8)"] $ \unterminated ->+    assert "unterminated SQL construct" (isDefinitionError UnterminatedSql (sqlStep "x" unterminated))+  forM_ ["BEGIN", "COMMIT", "ROLLBACK", "USE app", "SET x=1", "SELECT 1", "SYSTEM FLUSH LOGS", "CREATE TABLE app.t ON /* gap */ CLUSTER c (x UInt8)", "INSERT INTO app.t FORMAT CSV\n1", "CREATE MATERIALIZED VIEW app.v REFRESH EVERY 1 SECOND AS SELECT 1", "ALTER TABLE app.t MOVE PARTITION 1 TO TABLE app.other", "CREATE TABLE app.t ENGINE=ReplicatedMergeTree ORDER BY tuple()", "CREATE TABLE unqualified (x UInt8)", "\\i file.sql"] $ \unsupported ->+    assert "unsupported SQL" (isDefinitionError UnsupportedSql (sqlStep "x" unsupported))+  forM_ ["async_insert", "mutations_sync", "alter_sync", "AsYnC_InSeRt", "`alter_sync`", "\"mutations_sync\""] $ \setting ->+    assert "protected setting" (isDefinitionError ProtectedSetting (sqlStep "x" (Encoding.encodeUtf8 ("INSERT INTO app.t VALUES (1) SETTINGS " <> setting <> " /*comment*/ = 0"))))+  forM_ ["INSERT INTO app.t VALUES ('a; b')", "INSERT INTO app.t VALUES ('a'';b')", "INSERT INTO app.t VALUES ('a\\';b')", "INSERT INTO app.t VALUES ($tag$a; ' \\ $tag$)", "INSERT INTO app.t VALUES ($$a; b$$)", "CREATE TABLE app.\"t;\" (`x;` UInt8)", "CREATE TABLE app.\"t\\\";\" (x UInt8)", "CREATE TABLE app.t (x UInt8) /* outer; /* inner; */ end; */; -- tail;", "INSERT INTO app.t VALUES (1) // ; ignored\n; #! tail;", "ALTER TABLE app.t ADD COLUMN x UInt8", "ALTER TABLE app.t UPDATE x=1 WHERE x=0", "INSERT INTO app.t SELECT 1", "DROP TABLE IF EXISTS app.t", "CREATE DATABASE app", "CREATE MATERIALIZED VIEW app.v TO app.t AS SELECT * FROM app.s"] $ \valid -> do+    step <- right (sqlStep "x" valid)+    m <- right (sqlMigration "m" (step :| []))+    c <- right (migrationComponent "c" Set.empty (m :| []))+    p <- right (migrationPlan (c :| []))+    assert "SQL exact bytes preserved" (stepSql (NE.head (migrationSteps (NE.head (componentMigrations (NE.head (inspectPlan p)))))) == valid)+  -- Boundary combinations cover escaped delimiters, comment forms and heredoc tags.+  forM_ [";", "'; DROP TABLE app.t; --", "/* ; */", "雪;", "\\;", "# ;"] $ \payload ->+    forM_ ["tag", "other", "_x9"] $ \tag -> do+      _ <- right (sqlStep "x" (Encoding.encodeUtf8 ("INSERT INTO app.t VALUES ($" <> tag <> "$" <> payload <> "$" <> tag <> "$); /* ; */")))+      pure ()+  described <- right (migrationComponent "a" Set.empty (withMigrationDescription (Just "metadata") migration :| []))+  describedPlan <- right (migrationPlan (described :| []))+  let firstMigration = NE.head . componentMigrations . NE.head . inspectPlan+  assert "description excluded from checksum" (migrationChecksum (firstMigration plan) == migrationChecksum (firstMigration describedPlan))+  assert "description available" (migrationDescription (firstMigration describedPlan) == Just "metadata")+  assert "unknown checksum format" (isDefinitionError UnknownChecksumFormat (parseChecksumFormat "sha256-v2"))+  assert "known checksum format" (parseChecksumFormat "sha256-v1" == Right ChecksumV1)+  vectorLines <- filter (not . Text.isPrefixOf "#") . Text.lines <$> Text.readFile "test/vectors/checksum-v1.tsv"+  forM_ vectorLines checkVector+  putStrLn "Definition, ordering, SQL boundary and 24 independent checksum-vector checks passed."++checkVector :: Text -> IO ()+checkVector line = case Text.splitOn "\t" line of+  [label, kind, component, name, sources, expectedInput, expectedDigest] -> do+    steps <- mapM parseStep (Text.splitOn "," sources)+    let encodedSteps = fmap (\(key, sql) -> Checksum.encodeStepV1 component name key sql) (NE.fromList steps)+        input = if kind == "step" then NE.head encodedSteps else Checksum.encodeMigrationV1 component name (NE.toList encodedSteps)+    expected <- unhex expectedInput+    assert (Text.unpack (label <> " encoded bytes")) (input == expected)+    assert (Text.unpack (label <> " SHA-256")) (Checksum.digestHex (Checksum.digest input) == expectedDigest)+    authored <- mapM (\(key, sql) -> right (sqlStep key sql)) steps+    m <- right (sqlMigrationFromList name authored)+    c <- right (migrationComponent component Set.empty (m :| []))+    p <- right (migrationPlan (c :| []))+    let bound = NE.head (componentMigrations (NE.head (inspectPlan p)))+        actual = if kind == "step" then stepChecksumHex (stepChecksum (NE.head (migrationSteps bound))) else migrationChecksumHex (migrationChecksum bound)+    assert (Text.unpack (label <> " public bound checksum")) (actual == expectedDigest)+  _ -> fail "Invalid checked-in checksum vector"+  where+    parseStep source = case Text.splitOn ":" source of+      [key, hex] -> (key,) <$> unhex hex+      _ -> fail "Invalid step in checksum vector"++unhex :: Text -> IO ByteString+unhex text = BS.pack <$> go (Text.unpack text)+  where+    go [] = pure []+    go (a : b : rest) = case readHex [a, b] of+      [(byte, "")] -> (byte :) <$> go rest+      _ -> fail "Invalid vector hex"+    go _ = fail "Odd vector hex length"++-- Public parsing boundaries must reject URL credentials and reserved overrides.+targetValidation :: IO ()+targetValidation = do+  forM_ ["http://localhost:8123", "https://example.com/", "http://[::1]:8123"] $ \value -> do+    _ <- right (directEndpoint value)+    pure ()+  forM_ ["", "ftp://localhost", "http://user:secret@localhost", "http://localhost/path", "http://localhost?async_insert=1", "http://localhost#fragment", "http://localhost:0", "http://localhost:65536", "http://localhost//", "http://"] $ \value ->+    assert "reject unsafe endpoint" (directEndpoint value == Left InvalidEndpoint)+  forM_ ["00000000-0000-0000-0000-000000000000", "01234567-89ab-cdef-0123-456789abcdef"] $ \value -> do+    _ <- right (serverId value)+    _ <- right (ledgerId value)+    pure ()+  forM_ ["", "01234567-89AB-cdef-0123-456789abcdef", "0123456789ab-cdef-0123-456789abcdef"] $ \value ->+    assert "canonical UUID required" (serverId value == Left InvalidServerId)+  assert "invalid request bounds" (either (== InvalidHttpLimits) (const False) (httpLimits 0 1))+  assert "bounded response allocation" (either (== InvalidHttpLimits) (const False) (httpLimits 1 16777217))+  assert "credential header injection" (either (== InvalidCredentials) (const False) (basicCredentials "user\r\nheader" "password"))+  assert "namespace cannot inject SQL" (ledgerNamespace "ledger; DROP DATABASE app" == Left InvalidLedgerNamespace)++-- A library may append its own migration after another component has activated.+scheduleValidation :: IO ()+scheduleValidation = do+  step <- right (sqlStep "x" "CREATE TABLE app.x (id UInt8) ENGINE=Memory")+  one <- right (sqlMigration "one" (step :| []))+  two <- right (sqlMigration "two" (step :| []))+  a <- right (migrationComponent "a" Set.empty (one :| [two]))+  b <- right (migrationComponent "b" (Set.singleton "a") (one :| []))+  let ident c m = migrationIdFromNames <$> componentName c <*> migrationName m+  a1 <- right (ident "a" "one")+  a2 <- right (ident "a" "two")+  b1 <- right (ident "b" "one")+  unknown <- right (ident "a" "missing")+  explicit <- right (scheduledMigrationPlan (a :| [b]) (a1 :| [b1, a2]))+  ordinary <- right (migrationPlan (a :| [b]))+  frozen <- BS.readFile "test/vectors/schedule-default-v1.json"+  assert "frozen ordinary descriptors/checksums" (Codec.encodeValue (Aeson.toJSON (map Codec.definitionValue (Codec.definitions ordinary))) <> "\n" == frozen)+  assert "explicit order" (map migrationId (NE.toList (inspectSchedule explicit)) == [a1, b1, a2])+  assert "schedule keeps catalog and checksum views" (inspectPlan explicit == inspectPlan ordinary)+  forM_ [(MissingScheduledMigration, [a1, b1]), (UnknownScheduledMigration, [a1, b1, unknown]), (DuplicateScheduledMigration, [a1, b1, a2, a1]), (MigrationOrderChanged, [a2, a1, b1]), (DependencyNotActivated, [b1, a1, a2])] $ \(code, ids) ->+    assert "invalid schedule" (isPlanError code (scheduledMigrationPlanFromList [a, b] ids))+  assert "empty schedule" (isPlanError EmptySchedule (scheduledMigrationPlanFromList [a, b] []))++streamValidation :: IO ()+streamValidation = do+  let run bound chunks = do+        ref <- newIORef (filter (not . BS.null) chunks)+        readJsonRowStream bound (atomicModifyIORef' ref (\case [] -> ([], ""); x : xs -> (xs, x)))+      bytes = Encoding.encodeUtf8 "{\"value\":\"雪\\n\\\"\"}"+      success count = \case Transport.AcknowledgedSuccess (Transport.Rows rows) -> length rows == count; _ -> False+  expected <- run 1024 [bytes]+  forM_ [0 .. BS.length bytes] $ \i -> do+    actual <- run 1024 [BS.take i bytes, BS.drop i bytes]+    assert "every row byte split" (actual == expected)+  forM_ [[bytes <> "\n" <> bytes <> "\n"], [" \t\n", bytes, "\r\n", bytes], map BS.singleton (BS.unpack (bytes <> "\n" <> bytes))] $ \chunks ->+    assert "stream multiple/blank/CRLF rows" . success 2 =<< run 1024 chunks+  assert "exact row bound" . success 1 =<< run (BS.length bytes) [bytes <> "\n"]+  assert "no aggregate byte bound" . success 2 =<< run (BS.length bytes) [bytes <> "\n" <> bytes]+  invalidUtf8 <- run 1024 [bytes <> "\n{\"value\":\"" <> BS.singleton 255 <> "\"}"]+  assert "invalid row UTF-8 discards prefix" (case invalidUtf8 of Transport.SubmissionUncertain (Transport.TransportError Transport.UnexpectedResponse _ _) -> True; _ -> False)+  oversized <- run (BS.length bytes - 1) [bytes]+  assert "row bound plus one" (case oversized of Transport.SubmissionUncertain (Transport.TransportError Transport.ResponseTooLarge _ _) -> True; _ -> False)+  partial <- run 1024 [bytes <> "\n{\"value\":"]+  assert "no partial final row" (not (success 1 partial))+  late <- run 1024 [bytes <> "\nCode: 999. synthetic failure\n"]+  assert "late error discards rows" (case late of Transport.ObservedServerFailure (Transport.TransportError Transport.ServerRejected _ (Just 999)) -> True; _ -> False)+  assert "empty stream" . success 0 =<< run 1024 []
+ test/public/AdoptionConstructor.hs view
@@ -0,0 +1,5 @@+module Main where++import ClickHouseMigrate.Adoption++main = print (AdoptionPlan undefined undefined undefined)
+ test/public/Coercion.hs view
@@ -0,0 +1,8 @@+module Coercion where++import ClickHouseMigrate+import Data.Coerce (coerce)+import Data.Text (Text)++forge :: Text -> ComponentName+forge = coerce
+ test/public/Constructor.hs view
@@ -0,0 +1,6 @@+module Constructor where++import ClickHouseMigrate++forge :: StepName -> SqlStep+forge name = SqlStep name mempty
+ test/public/Deserialize.hs view
@@ -0,0 +1,6 @@+module Deserialize where++import ClickHouseMigrate++forge :: String -> SqlStep+forge = read
+ test/public/Generic.hs view
@@ -0,0 +1,7 @@+module Generic where++import ClickHouseMigrate+import GHC.Generics (Generic, Rep, to)++forge :: Rep MigrationPlan x -> MigrationPlan+forge = to
+ test/public/LedgerConstructor.hs view
@@ -0,0 +1,9 @@+module Main where++import ClickHouseMigrate.Ledger++bad :: LedgerPosition+bad = LedgerPosition undefined 1 undefined++main :: IO ()+main = pure ()
+ test/public/MissingProvider.hs view
@@ -0,0 +1,10 @@+module Main where++import ClickHouseMigrate.Ledger+import ClickHouseMigrate.Target++bad :: TargetConfig -> InitializationAuthorization -> QuiescenceEvidence -> IO (LedgerCommandResult LedgerPosition)+bad config authorization quiet = initializeLedger config authorization quiet++main :: IO ()+main = pure ()
+ test/public/MissingQuiescence.hs view
@@ -0,0 +1,11 @@+module Main where++import ClickHouseMigrate.ExclusiveAccess+import ClickHouseMigrate.Ledger+import ClickHouseMigrate.Target++bad :: TargetConfig -> ExclusiveAccess -> InitializationAuthorization -> IO (LedgerCommandResult LedgerPosition)+bad config provider authorization = initializeLedger config provider authorization++main :: IO ()+main = pure ()
+ test/public/Optic.hs view
@@ -0,0 +1,11 @@+module Optic where++import ClickHouseMigrate+import Data.ByteString (ByteString)+import Data.Functor.Identity (Identity (..))++set :: ((a -> Identity a) -> s -> Identity s) -> a -> s -> s+set optic value source = runIdentity (optic (const (Identity value)) source)++forge :: ByteString -> StepView -> StepView+forge = set stepSql
+ test/public/Private.hs view
@@ -0,0 +1,3 @@+module Private where++import ClickHouseMigrate.Internal.Types
+ test/public/PrivateLedger.hs view
@@ -0,0 +1,6 @@+module Main where++import ClickHouseMigrate.Internal.LedgerStore++main :: IO ()+main = pure ()
+ test/public/RepairConstructor.hs view
@@ -0,0 +1,9 @@+module Main where++import ClickHouseMigrate.Repair++bad :: RepairRequest+bad = ResumeRequest undefined undefined undefined++main :: IO ()+main = pure ()
+ test/public/RuntimeConstructor.hs view
@@ -0,0 +1,9 @@+module Main where++import ClickHouseMigrate.Runner++bad :: MutationContext+bad = MutationContext undefined undefined undefined undefined++main :: IO ()+main = pure ()
+ test/public/SourceChecksumConstructor.hs view
@@ -0,0 +1,5 @@+module Main where++import ClickHouseMigrate.Adoption++main = print (SourceChecksum "sha256" "unchecked")
+ test/public/Valid.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import ClickHouseMigrate+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set++libraryA :: Either DefinitionError MigrationComponent+libraryA = do+  step <- sqlStep "create" "CREATE TABLE app.a (id UInt64) ENGINE=MergeTree ORDER BY id"+  migration <- sqlMigration "0001" (step :| [])+  migrationComponent "library-a" Set.empty (withMigrationDescription (Just "A") migration :| [])++libraryB :: Either DefinitionError MigrationComponent+libraryB = do+  step <- sqlStep "create" "CREATE TABLE app.b (id UInt64) ENGINE=MergeTree ORDER BY id"+  migration <- sqlMigration "0001" (step :| [])+  migrationComponent "library-b" (Set.singleton "library-a") (migration :| [])++main :: IO ()+main = case (libraryA, libraryB) of+  (Right a, Right b) -> case do+    ca <- componentName "library-a"+    cb <- componentName "library-b"+    name <- migrationName "0001"+    pure [migrationIdFromNames ca name, migrationIdFromNames cb name] of+    Right ids -> case scheduledMigrationPlanFromList [a, b] ids of+      Right p+        | map (componentNameText . componentViewName) (NE.toList (inspectPlan p)) == ["library-a", "library-b"] ->+            mapM_ (print . migrationChecksumHex . migrationChecksum) (concatMap (NE.toList . componentMigrations) (NE.toList (inspectPlan p)))+      _ -> fail "Public plan composition failed"+    _ -> fail "Public schedule identity failed"+  _ -> fail "Public authoring failed"
+ test/vectors/adoption-v1.json view
@@ -0,0 +1,20 @@+[+  {+    "level": "operator-attestation",+    "snapshotInputHex": "636c69636b686f7573652d6d6967726174652f61646f7074696f6e2d736e617073686f7400763100000000000000000e6c65676163792d66697874757265000000000000001a666978747572653a696d6d757461626c652d736e617073686f74000000000000000100000000000000073030312e73716c0000000000000014323032302d30312d30315430303a30303a30305a0000000000000000",+    "snapshot": "0cbd644a1b1871101f5944d249d1d48e6f83f487e43a427eced729390ba9e8f1",+    "mappingInputHex": "636c69636b686f7573652d6d6967726174652f61646f7074696f6e2d6d617070696e6700763100000000000000004030636264363434613162313837313130316635393434643234396431643438653666383366343837653433613432376563656437323933393062613965386631000000000000000100000000000000073030312e73716c000000000000010e7b22636865636b73756d223a2262646439396264653164333965386632373135653337643434373739376637393334356637343064373434306436383739323764613861373632393035326539222c22636f6d706f6e656e74223a22617070222c22636f6d706f6e656e74496e646578223a302c22646570656e64656e63696573223a5b5d2c226d6967726174696f6e223a226f6e65222c226f7264696e616c223a302c227374657073223a5b7b22636865636b73756d223a2237373163396665313337303237373663376664636534616233633937663961343966323738363138313135356565373664386132623935306666313933653461222c226e616d65223a22637265617465227d5d7d000000000000000000000000000000040000000000000015736f757263653a6c65676163792d66697874757265000000000000002c736e617073686f742d65766964656e63653a666978747572653a696d6d757461626c652d736e617073686f740000000000000012736f757263652d726f773a3030312e73716c000000000000001d666978747572653a72657669657765642d737465702d6d617070696e67",+    "mapping": "12a26e9bb47128906ec2717968071b3b32ab931670ba83fbf4a8475b27394408",+    "definitionsInputHex": "5b7b22636865636b73756d223a2262646439396264653164333965386632373135653337643434373739376637393334356637343064373434306436383739323764613861373632393035326539222c22636f6d706f6e656e74223a22617070222c22636f6d706f6e656e74496e646578223a302c22646570656e64656e63696573223a5b5d2c226d6967726174696f6e223a226f6e65222c226f7264696e616c223a302c227374657073223a5b7b22636865636b73756d223a2237373163396665313337303237373663376664636534616233633937663961343966323738363138313135356565373664386132623935306666313933653461222c226e616d65223a22637265617465227d5d7d5d",+    "definitions": "efa03f51657e440641d3a2a303eebc9b0a933fac99f8ee1e4270e0d13416707c"+  },+  {+    "level": "predecessor-checksum",+    "snapshotInputHex": "636c69636b686f7573652d6d6967726174652f61646f7074696f6e2d736e617073686f7400763100000000000000000e6c65676163792d66697874757265000000000000001a666978747572653a696d6d757461626c652d736e617073686f74000000000000000100000000000000073030312e73716c0000000000000014323032302d30312d30315430303a30303a30305a00000000000000010000000000000006736861323536000000000000004035306661626364653263363034336539346435646664633033616435383766323237616363663031383661646162626538656237353336623331313433333538",+    "snapshot": "4906c7532c886734513805a2c730dcb0743e1e7a74ec621965a2b509985213bf",+    "mappingInputHex": "636c69636b686f7573652d6d6967726174652f61646f7074696f6e2d6d617070696e6700763100000000000000004034393036633735333263383836373334353133383035613263373330646362303734336531653761373465633632313936356132623530393938353231336266000000000000000100000000000000073030312e73716c000000000000010e7b22636865636b73756d223a2262646439396264653164333965386632373135653337643434373739376637393334356637343064373434306436383739323764613861373632393035326539222c22636f6d706f6e656e74223a22617070222c22636f6d706f6e656e74496e646578223a302c22646570656e64656e63696573223a5b5d2c226d6967726174696f6e223a226f6e65222c226f7264696e616c223a302c227374657073223a5b7b22636865636b73756d223a2237373163396665313337303237373663376664636534616233633937663961343966323738363138313135356565373664386132623935306666313933653461222c226e616d65223a22637265617465227d5d7d000000000000000100000000000000060000000000000015736f757263653a6c65676163792d66697874757265000000000000002c736e617073686f742d65766964656e63653a666978747572653a696d6d757461626c652d736e617073686f740000000000000012736f757263652d726f773a3030312e73716c000000000000001b686973746f726963616c2d616c676f726974686d3a7368613235360000000000000054686973746f726963616c2d636865636b73756d3a35306661626364653263363034336539346435646664633033616435383766323237616363663031383661646162626538656237353336623331313433333538000000000000001c666978747572653a7072656465636573736f722d636865636b73756d",+    "mapping": "32c3c3153d39088574f46249fa000b028e4c8353c9e04e6df7d49c054ecb5052",+    "definitionsInputHex": "5b7b22636865636b73756d223a2262646439396264653164333965386632373135653337643434373739376637393334356637343064373434306436383739323764613861373632393035326539222c22636f6d706f6e656e74223a22617070222c22636f6d706f6e656e74496e646578223a302c22646570656e64656e63696573223a5b5d2c226d6967726174696f6e223a226f6e65222c226f7264696e616c223a302c227374657073223a5b7b22636865636b73756d223a2237373163396665313337303237373663376664636534616233633937663961343966323738363138313135356565373664386132623935306666313933653461222c226e616d65223a22637265617465227d5d7d5d",+    "definitions": "efa03f51657e440641d3a2a303eebc9b0a933fac99f8ee1e4270e0d13416707c"+  }+]
+ test/vectors/checksum-v1.tsv view
@@ -0,0 +1,25 @@+# label	kind	component	migration	steps(name:sql-hex,...)	input-hex	sha256-hex+baseline	step	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	03f833fc579162357199f7aba650e61b1af2bc8954f4a0ff1d15b6a81f228189+baseline	migration	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000100000000000000e3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	89282ba04db62ba8892ca003e3cce3a3d095538ebff1dc326362de6721e6ef19+whitespace	step	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000041435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	7f09e988732212202859fcf53c33e24e4e9373488f9e915fe9c65e2669333648+whitespace	migration	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000100000000000000e4636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000041435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	ac0fb42387103d070ab0823b79d5791a5c556a38dec1b8a0a2280f3a35588fab+component	step	analytics2	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d31000000000000000a616e616c79746963733200000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	8f6e63969a02861056085a9aff5d0f8f312e77c4659117a2fc2e35b2de4d92ce+component	migration	analytics2	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d31000000000000000a616e616c797469637332000000000000000430303031000000000000000100000000000000e4636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d31000000000000000a616e616c79746963733200000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	8be828d6c9ee3f333b5efe1b55e0664df45fd2c9d1160938d4066ca57ae49dda+migration	step	analytics	0002	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303200000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	02fae0c23bb7624f887f00412bde13dae9ef7bf2a63fec82bacaabf54047d8cd+migration	migration	analytics	0002	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303032000000000000000100000000000000e3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303200000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	055a0739a1e204f3451a552dfd23e8ce38609db113adb04260d1e1c6418920cc+step-name	step	analytics	0001	create2:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000007637265617465320000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	5aea7c57971279de316f3df5044be50a2e505abe9eec86aa4a7b9e510bcf07d6+step-name	migration	analytics	0001	create2:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000100000000000000e4636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000007637265617465320000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	08e1b92eec60b42d7d09ce56d53199a0d78a68790944b0c95ea80ac3f02b60dd+two-steps	step	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	03f833fc579162357199f7aba650e61b1af2bc8954f4a0ff1d15b6a81f228189+two-steps	migration	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964,seed:494e5345525420494e544f206170702e6576656e74732056414c55455320283129	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000200000000000000e3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f5244455220425920696400000000000000c2636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000021494e5345525420494e544f206170702e6576656e74732056414c55455320283129	04e037c3b68aef7f8549610f53b4e229854e985dc5a2f742f4c3a8a8b849edc2+reordered	step	analytics	0001	seed:494e5345525420494e544f206170702e6576656e74732056414c55455320283129	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000021494e5345525420494e544f206170702e6576656e74732056414c55455320283129	8bd18e0b06f44a08db320baa745b9fd9df910da6957f9436b2be574b4acaabaf+reordered	migration	analytics	0001	seed:494e5345525420494e544f206170702e6576656e74732056414c55455320283129,create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000200000000000000c2636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000021494e5345525420494e544f206170702e6576656e74732056414c5545532028312900000000000000e3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	14d70d5123a928b2a3bf414b69dffbd0708889187fd217ada33330365961147a+boundary-a	step	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000041435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a	7f09e988732212202859fcf53c33e24e4e9373488f9e915fe9c65e2669333648+boundary-a	migration	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a,seed:494e5345525420494e544f206170702e6576656e74732056414c55455320283129	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000200000000000000e4636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000041435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f524445522042592069640a00000000000000c2636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000021494e5345525420494e544f206170702e6576656e74732056414c55455320283129	077c5b23428a0c62d9d6296b95a92cd7ddd2894f2d1aaf82e686bfa15953eb96+boundary-b	step	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	03f833fc579162357199f7aba650e61b1af2bc8954f4a0ff1d15b6a81f228189+boundary-b	migration	analytics	0001	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964,seed:0a494e5345525420494e544f206170702e6576656e74732056414c55455320283129	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000200000000000000e3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f5244455220425920696400000000000000c3636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c797469637300000000000000043030303100000000000000047365656400000000000000220a494e5345525420494e544f206170702e6576656e74732056414c55455320283129	88c15fe35e78ba1f38923ab8a28229a71a2e64269a3998acc2b27047e77d221e+ambiguous-a	step	a	bc	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000001610000000000000002626300000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	07bd3b2d6b887eaf8f5807fb557ce25d42361e7f3529115a8d192794d316484e+ambiguous-a	migration	a	bc	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d3100000000000000016100000000000000026263000000000000000100000000000000d9636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000001610000000000000002626300000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	8e184656b6d8165aa19508e32f7e5311c1424a9ede170e79bb46f3fcaa7904eb+ambiguous-b	step	ab	c	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000002616200000000000000016300000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	de15311def4cde65b9d774b9dd5207c2f8129f1d19290373c5e572600bc6c262+ambiguous-b	migration	ab	c	create:435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d3100000000000000026162000000000000000163000000000000000100000000000000d9636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000002616200000000000000016300000000000000066372656174650000000000000040435245415445205441424c45206170702e6576656e7473202869642055496e7436342920454e47494e453d4d6572676554726565204f52444552204259206964	f98499b7805b491bd9271c9c69251f0b9d31ef00c9a45abc9d4e96877e20d081+unicode	step	analytics	0001	seed:494e5345525420494e544f206170702e746578742056414c554553202827e99baa3b2729	636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000024494e5345525420494e544f206170702e746578742056414c554553202827e99baa3b2729	bbadfbf575dda692e57a822702bf48b12c32d33ceb8e5ace531ad4bd78f33e67+unicode	migration	analytics	0001	seed:494e5345525420494e544f206170702e746578742056414c554553202827e99baa3b2729	636c69636b686f7573652d6d6967726174652f6d6967726174696f6e00763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c7974696373000000000000000430303031000000000000000100000000000000c5636c69636b686f7573652d6d6967726174652f7374657000763100000000000000004d636c69636b686f7573652d6d6967726174652f6c6f63616c2d73796e632d76313b6173796e635f696e736572743d303b6d75746174696f6e735f73796e633d313b616c7465725f73796e633d310000000000000009616e616c79746963730000000000000004303030310000000000000004736565640000000000000024494e5345525420494e544f206170702e746578742056414c554553202827e99baa3b2729	31169fd8dc8276b08d2ca8a664dd3530a92c58d332a5b13af63c5c1a99077b92
+ test/vectors/ledger-initialized-v1.json view
@@ -0,0 +1,1 @@+{"attemptId":null,"audit":{"actor":"operator's \"review\"","reason":"test\nreason","references":["mori://shinzui/clickhouse-migrate"]},"body":{"checksumFormat":"sha256-v1","policy":"clickhouse-migrate/local-sync-v1;async_insert=0;mutations_sync=1;alter_sync=1","schemaVersion":1,"target":"development/main"},"context":null,"eventId":"00000000-0000-4000-8000-000000000001","kind":"initialized","ledgerId":"00000000-0000-4000-8000-000000000099","occurredAt":"2026-10-16T00:00:00Z","runId":"00000000-0000-4000-8000-000000000098","runnerVersion":"0.1.0.0","sequence":1,"version":1}
+ test/vectors/schedule-default-v1.json view
@@ -0,0 +1,1 @@+[{"checksum":"afde71d4ba89a273e9951b2983c305d12984879121b543c9c7c309ea07575e60","component":"a","componentIndex":0,"dependencies":[],"migration":"one","ordinal":0,"steps":[{"checksum":"93c9e122f428248cf58a5df6a4a0c904423e55624e164dac374c8364c49c372b","name":"x"}]},{"checksum":"ee3d635e5742051fe39e56cf25a7e67a250cdbe7771206fb108f4b54a5edc4ad","component":"a","componentIndex":0,"dependencies":[],"migration":"two","ordinal":1,"steps":[{"checksum":"97c54c0d093be5c92673ceaf89771d796e71eac21dedd90262e4030e20ff2d7f","name":"x"}]},{"checksum":"2e8472fdb4223cc867ccc3fa49347661c40db60103ce7c9eaacd67492a816e51","component":"b","componentIndex":1,"dependencies":["a"],"migration":"one","ordinal":2,"steps":[{"checksum":"0936a18df42cb36ecf7a7fcf1e5f3b64615e49742fbc430af674f1254beb9d48","name":"x"}]}]