clickhouse-migrate 0.2.0.0 → 0.2.1.0

raw patch · 12 files changed

+116/−26 lines, 12 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +14/−0
• clickhouse-migrate.cabal +1/−1
• src/ClickHouseMigrate.hs +1/−0
• src/ClickHouseMigrate/Inspection.hs +14/−1
• src/ClickHouseMigrate/Internal/Http.hs +7/−1
• src/ClickHouseMigrate/Internal/RepairTypes.hs +14/−1
• src/ClickHouseMigrate/Internal/RuntimeTypes.hs +1/−1
• src/ClickHouseMigrate/Internal/TargetTypes.hs +2/−1
• src/ClickHouseMigrate/Repair.hs +55/−16
• src/ClickHouseMigrate/Target.hs +2/−2
• test-driver/Main.hs +2/−2
• test/Main.hs +3/−0

@@ -1,5 +1,19 @@ # Changelog for clickhouse-migrate +## 0.2.1.0 — 2026-11-14++### New Features++- Add `boundedObservation` for explicit bounded SELECT-only inspection by+  constrained reader roles. Overflow modes remain throwing, so an overflowing+  history fails rather than being truncated; `Observe` and mutation paths keep+  their existing settings.+- Add `externalDrainRecoveryEvidence` and typed `RepairObservation` callbacks+  (`RepairObservationSink`) for callers retaining independent external drain.+  Only explicit process access denial becomes unavailable diagnostics; visible+  recorded work and all ordinary repair checks remain blocking. Existing+  `recoveryEvidence` remains strict.+ ## 0.2.0.0 — 2026-11-08  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: clickhouse-migrate-version: 0.2.0.0+version: 0.2.1.0 synopsis: Core definitions and execution for ClickHouse migrations description:   Core definitions and execution for ClickHouse migrations.

@@ -11,6 +11,7 @@     upOptions,     up,     InspectionOptions (Observe, Guarded),+    boundedObservation,     VerifyOptions,     verifyOptions,     status,

@@ -1,6 +1,7 @@ -- | Read-only observations. No inspection path initializes or repairs history. module ClickHouseMigrate.Inspection   ( InspectionOptions (Observe, Guarded),+    boundedObservation,     VerifyOptions,     verifyOptions,     status,@@ -25,11 +26,22 @@ import Control.Exception (SomeAsyncException, SomeException, fromException, throwIO, try) import Data.IORef (newIORef, readIORef, writeIORef) +-- | An explicitly bounded SELECT-only observation for constrained reader roles.+-- Server overflow modes remain throwing, so a truncated history is never success.+-- No mutation or guarded-inspection path accepts these bounds.+boundedObservation :: Int -> Int -> Maybe InspectionOptions+boundedObservation rows bytes+  | rows > 0 && bytes > 0 = Just (ObserveBounded rows bytes)+  | otherwise = Nothing+ verifyOptions :: Bool -> VerifyOptions verifyOptions = VerifyOptions  history :: InspectionOptions -> TargetConfig -> IO (Either R.InspectionFailure R.HistoryReport)-history options config@(TargetConfig {configKey = target}) = do+history options original@(TargetConfig {configKey = target}) = do+  let config = case options of+        ObserveBounded rows bytes -> original {configObservationLimits = Just (rows, bytes)}+        _ -> original   latest <- newIORef Nothing   observed <- try @SomeException $ do     probe <- QueryId . ("chm-" <>) <$> freshUuid@@ -51,6 +63,7 @@           )     case options of       Observe -> flatten <$> scoped Nothing R.UnguardedObservation+      ObserveBounded _ _ -> 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

@@ -243,7 +243,7 @@   let request =         HTTP.applyBasicAuth user password $           HTTP.setQueryString-            (Profile.requestSettings (configTopology config) (configLimits config) identity kind)+            (observationSettings config kind (Profile.requestSettings (configTopology config) (configLimits config) identity kind))             base               { HTTP.method = "POST",                 HTTP.requestBody = HTTP.RequestBodyBS sql,@@ -260,3 +260,9 @@   identity <- myThreadId   atomicModifyIORef' threads (\old -> (identity : old, ()))   restore action `finally` atomicModifyIORef' threads (\old -> (filter (/= identity) old, ()))++-- Used only by Inspection.boundedObservation; defaults retain their byte pins.+observationSettings :: TargetConfig -> RequestKind -> [(ByteString, Maybe ByteString)] -> [(ByteString, Maybe ByteString)]+observationSettings config kind settings = case configObservationLimits config of+  Just (rows, bytes) | kind /= Command -> map (\(key, value) -> (key, if key == "max_result_rows" then Just (Encoding.encodeUtf8 (Text.pack (show rows))) else if key == "max_result_bytes" then Just (Encoding.encodeUtf8 (Text.pack (show bytes))) else value)) settings+  _ -> settings

@@ -2,8 +2,21 @@  import ClickHouseMigrate.Ledger (AppendReceipt, AttemptId, AuditContext, LedgerPosition, QuiescenceEvidence) import ClickHouseMigrate.Report (StepAttempt)+import ClickHouseMigrate.Target (QueryId) -data RecoveryEvidence = RecoveryEvidence !AuditContext !QuiescenceEvidence ![AppendReceipt]+-- Observations describe diagnostics, never the authority to enter recovery.+data RepairObservation+  = NoRecordedQueries+  | NoVisibleRecordedQueries+  | RecordedQueriesActive ![QueryId]+  | ProcessObservationUnavailable+  deriving stock (Eq, Show)++type RepairObservationSink = RepairObservation -> IO ()++data ProcessObservationPolicy = StrictObservation | ExternalDrain !RepairObservationSink++data RecoveryEvidence = RecoveryEvidence !AuditContext !QuiescenceEvidence ![AppendReceipt] !ProcessObservationPolicy  data RepairRequest   = ResumeRequest !LedgerPosition !AttemptId !RecoveryEvidence

@@ -10,6 +10,6 @@  data UpOptions = UpOptions !AuditContext !(Maybe InitializationAuthorization) -data InspectionOptions = Observe | Guarded !ExclusiveAccess+data InspectionOptions = Observe | ObserveBounded !Int !Int | Guarded !ExclusiveAccess  data VerifyOptions = VerifyOptions !Bool deriving stock (Eq, Show)

@@ -32,7 +32,8 @@     configNamespace :: !LedgerNamespace,     configBinding :: !LedgerBinding,     configCredentials :: !Credentials,-    configLimits :: !HttpLimits+    configLimits :: !HttpLimits,+    configObservationLimits :: !(Maybe (Int, Int))   }  data ServerIdentity = ServerIdentity !ServerId !Text !Text deriving stock (Eq, Show)

@@ -4,6 +4,9 @@     RecoveryEvidence,     RepairRequestError (..),     recoveryEvidence,+    externalDrainRecoveryEvidence,+    RepairObservation (..),+    RepairObservationSink,     resumeRequest,     retryStepRequest,     acceptStepRequest,@@ -26,11 +29,13 @@ import ClickHouseMigrate.Internal.Reporting qualified as Reporting import ClickHouseMigrate.Internal.RuntimeTypes (MutationContext (..)) 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 qualified as Target+import Control.Exception (SomeException, throwIO, try) import Data.List (findIndex) import Data.List.NonEmpty qualified as NE import Data.Set qualified as Set@@ -45,8 +50,18 @@   | 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)+  | otherwise = Right (RecoveryEvidence audit quiet receipts StrictObservation) +-- | Opt in only after independently establishing and retaining external drain+-- under the caller's exclusion. A denied process-list read is then unavailable+-- diagnostics, not an empty observation. The sink is called before authorization+-- and must retain that distinction. All other errors and positive observations+-- remain blocking. The existing 'recoveryEvidence' keeps its strict behavior.+externalDrainRecoveryEvidence :: Ledger.AuditContext -> Ledger.QuiescenceEvidence -> [Ledger.AppendReceipt] -> RepairObservationSink -> Either RepairRequestError RecoveryEvidence+externalDrainRecoveryEvidence audit quiet receipts sink = do+  _ <- recoveryEvidence audit quiet receipts+  pure (RecoveryEvidence audit quiet receipts (ExternalDrain sink))+ resumeRequest :: Ledger.LedgerPosition -> Ledger.AttemptId -> RecoveryEvidence -> Either RepairRequestError RepairRequest resumeRequest position attempt evidence = validateReceipts position evidence >> Right (ResumeRequest position attempt evidence) @@ -60,7 +75,7 @@ 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)+validateReceipts (L.LedgerPosition ledger _ _) (RecoveryEvidence _ _ receipts _)   | any (\(L.AppendReceipt _ actual _ _ _ _) -> actual /= ledger) receipts = Left ForeignLedgerReceipt   | otherwise = Right () @@ -76,7 +91,7 @@  repair :: MutationContext -> Plan.MigrationPlan -> RepairRequest -> IO (Either R.RunFailure R.RunReport) repair context@(MutationContext (T.TargetConfig {T.configKey = target, T.configBinding = binding}) _ _ _) plan request =-  let (expected@(L.LedgerPosition ledger _ _), requested, RecoveryEvidence audit quiet receipts) = requestParts request+  let (expected@(L.LedgerPosition ledger _ _), requested, RecoveryEvidence audit quiet receipts observationPolicy) = requestParts request       prerequisite = case binding of         T.FirstUse -> Left R.ExistingBindingRequired         T.ExistingLedger actual | actual /= ledger -> Left (R.LedgerFailed L.LedgerIdentityMismatch)@@ -92,8 +107,18 @@         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))+        observation <- activeQueries runtime queries observationPolicy+        case observationPolicy of+          StrictObservation -> pure ()+          ExternalDrain sink -> do+            delivered <- try @SomeException (sink observation)+            case delivered of+              Left exception | isAsync exception -> throwIO exception+              Left _ -> abort (R.CallbackFailed (R.EventCallback "RepairProcessObservation"))+              Right () -> pure ()+        case observation of+          RecordedQueriesActive active -> abort (R.PriorQueriesActive active)+          _ -> pure ()         migration <- case drop (L.ordinal definition) (orderedMigrations (L.projectionChecksumFormat projection) plan) of           (actual, migration) : _ | actual == definition -> pure migration           _ -> abort (R.LedgerFailed L.DefinitionMismatch)@@ -188,15 +213,29 @@   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+activeQueries :: Runtime -> [T.QueryId] -> ProcessObservationPolicy -> IO RepairObservation+activeQueries _ [] _ = pure NoRecordedQueries+activeQueries runtime queries policy = 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+  observed <- Store.storeResult (Store.readRows (Just (runtimeGuard runtime)) (runtimeTransport runtime) (Encoding.encodeUtf8 sql))+  case observed of+    Left failure | denied failure && external policy -> pure ProcessObservationUnavailable+    Left failure -> abort (Reporting.storeError failure)+    Right rows -> do+      active <-+        callStore $+          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+      pure (if null active then NoVisibleRecordedQueries else RecordedQueriesActive active)+  where+    external (ExternalDrain _) = True+    external _ = False+    -- A complete explicit access-denied response is the only optional failure.+    -- Guard loss, timeouts, malformed rows and unrelated failures remain fatal.+    denied (Store.RequestProblem (Transport.ObservedServerFailure (Transport.TransportError Transport.ServerRejected _ (Just 497))) Nothing) = True+    denied _ = False

@@ -111,7 +111,7 @@ defaultHttpLimits = HttpLimits 30000000 1048576  targetConfig :: TargetKey -> DirectEndpoint -> ServerId -> LedgerNamespace -> LedgerBinding -> Credentials -> HttpLimits -> TargetConfig-targetConfig key endpoint identity = TargetConfig key endpoint (DirectServer identity)+targetConfig key endpoint identity namespace binding credentials limits = TargetConfig key endpoint (DirectServer identity) namespace binding credentials limits Nothing  targetKeyText :: TargetKey -> Text targetKeyText (TargetKey value) = value@@ -183,4 +183,4 @@ replicatedTargetConfig key endpoint cluster replicas namespace binding credentials limits@(HttpLimits micros _)   | Set.size (Set.fromList (NE.toList replicas)) /= NE.length replicas = Left DuplicateReplica   | micros < 5000000 = Left InvalidHttpLimits-  | otherwise = Right (TargetConfig key endpoint (ReplicatedCluster cluster replicas) namespace binding credentials limits)+  | otherwise = Right (TargetConfig key endpoint (ReplicatedCluster cluster replicas) namespace binding credentials limits Nothing)

@@ -146,7 +146,7 @@ -- 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+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@@ -178,7 +178,7 @@     unlessTest condition message = if condition then pure () else fail message  ledgerProtocol :: String -> TargetConfig -> QueryId -> IO ()-ledgerProtocol mode config@(TargetTypes.TargetConfig key _ _ namespace _ _ _) probe = do+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"

@@ -48,6 +48,9 @@   streamValidation   scheduleValidation   targetValidation+  assert "zero row observation bound refused" (case boundedObservation 0 1 of Nothing -> True; _ -> False)+  assert "zero byte observation bound refused" (case boundedObservation 1 0 of Nothing -> True; _ -> False)+  assert "positive observation bounds admitted" (case boundedObservation 10000 1048576 of Just _ -> True; _ -> False)   directProfilePins   replicatedTargetValidation   let sql = "CREATE TABLE app.events (id UInt64) ENGINE=MergeTree ORDER BY id"