diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/clickhouse-migrate.cabal b/clickhouse-migrate.cabal --- a/clickhouse-migrate.cabal +++ b/clickhouse-migrate.cabal @@ -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. diff --git a/src/ClickHouseMigrate.hs b/src/ClickHouseMigrate.hs --- a/src/ClickHouseMigrate.hs +++ b/src/ClickHouseMigrate.hs @@ -11,6 +11,7 @@ upOptions, up, InspectionOptions (Observe, Guarded), + boundedObservation, VerifyOptions, verifyOptions, status, diff --git a/src/ClickHouseMigrate/Inspection.hs b/src/ClickHouseMigrate/Inspection.hs --- a/src/ClickHouseMigrate/Inspection.hs +++ b/src/ClickHouseMigrate/Inspection.hs @@ -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 diff --git a/src/ClickHouseMigrate/Internal/Http.hs b/src/ClickHouseMigrate/Internal/Http.hs --- a/src/ClickHouseMigrate/Internal/Http.hs +++ b/src/ClickHouseMigrate/Internal/Http.hs @@ -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 diff --git a/src/ClickHouseMigrate/Internal/RepairTypes.hs b/src/ClickHouseMigrate/Internal/RepairTypes.hs --- a/src/ClickHouseMigrate/Internal/RepairTypes.hs +++ b/src/ClickHouseMigrate/Internal/RepairTypes.hs @@ -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 diff --git a/src/ClickHouseMigrate/Internal/RuntimeTypes.hs b/src/ClickHouseMigrate/Internal/RuntimeTypes.hs --- a/src/ClickHouseMigrate/Internal/RuntimeTypes.hs +++ b/src/ClickHouseMigrate/Internal/RuntimeTypes.hs @@ -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) diff --git a/src/ClickHouseMigrate/Internal/TargetTypes.hs b/src/ClickHouseMigrate/Internal/TargetTypes.hs --- a/src/ClickHouseMigrate/Internal/TargetTypes.hs +++ b/src/ClickHouseMigrate/Internal/TargetTypes.hs @@ -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) diff --git a/src/ClickHouseMigrate/Repair.hs b/src/ClickHouseMigrate/Repair.hs --- a/src/ClickHouseMigrate/Repair.hs +++ b/src/ClickHouseMigrate/Repair.hs @@ -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 diff --git a/src/ClickHouseMigrate/Target.hs b/src/ClickHouseMigrate/Target.hs --- a/src/ClickHouseMigrate/Target.hs +++ b/src/ClickHouseMigrate/Target.hs @@ -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) diff --git a/test-driver/Main.hs b/test-driver/Main.hs --- a/test-driver/Main.hs +++ b/test-driver/Main.hs @@ -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" diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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"