clickhouse-migrate-test-support 0.1.0.0 → 0.2.0.0
raw patch · 9 files changed
+693/−56 lines, 9 filesdep ~clickhouse-migratedep ~clickhouse-migrate-clidep ~clickhouse-migrate-embedPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: clickhouse-migrate, clickhouse-migrate-cli, clickhouse-migrate-embed
API changes (from Hackage documentation)
+ ClickHouseMigrate.TestSupport.Server: queryReplica :: Server -> Int -> Text -> IO Value
+ ClickHouseMigrate.TestSupport.Server: startReplica :: Server -> Int -> IO ()
+ ClickHouseMigrate.TestSupport.Server: stopReplica :: Server -> Int -> IO ()
+ ClickHouseMigrate.TestSupport.Server: withReplicatedCandidate :: Text -> (Server -> IO a) -> IO a
Files
- CHANGELOG.md +19/−0
- clickhouse-migrate-test-support.cabal +6/−4
- src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs +9/−5
- src/ClickHouseMigrate/TestSupport/Server.hs +18/−0
- test/Main.hs +26/−18
- test/ReplicatedIntegration.hs +181/−0
- test/data/clickhouse-matrix.json +131/−3
- test/scripts/native_fixture.py +108/−26
- test/scripts/replicated_fixture.py +195/−0
CHANGELOG.md view
@@ -1,5 +1,24 @@ # Changelog for clickhouse-migrate-test-support +## 0.2.0.0 — 2026-11-08++### New Features++- Add owned Linux arm64 OCI fixtures for exact Altinity+ `26.3.60.10001.altinitystable` and `26.3.60.10002.altinitystable` engines.+- Add Keeper/two-replica fixtures and qualify replicated CLI execution, ledger+ storage, restarts, membership rejection, uncertainty and audited recovery.+- Preserve fixture identity and stable endpoints across owned server restarts.++### Bug Fixes++- Wait for writable database and table replicas during restart readiness.++### Other Changes++- Declare candidate-specific matrix groups and share them across artifact variants.+- Require the 0.2 series of the core, CLI and embedding libraries.+ ## 0.1.0.0 — 2026-10-16 Initial release.
clickhouse-migrate-test-support.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: clickhouse-migrate-test-support-version: 0.1.0.0+version: 0.2.0.0 synopsis: Isolated server fixtures for ClickHouse migration tests description: Isolated server fixtures for ClickHouse migration tests.@@ -22,6 +22,7 @@ test/fixtures/kansoku/*.json test/fixtures/kansoku/original/migrate.sh test/scripts/native_fixture.py+ test/scripts/replicated_fixture.py extra-source-files: test/scripts/transport_smoke.py@@ -44,7 +45,7 @@ ghc-options: -Wall build-depends: base >=4.21 && <4.22,- clickhouse-migrate ==0.1.*,+ clickhouse-migrate ==0.2.*, build-depends: aeson >=2.2.4.1 && <2.4,@@ -91,6 +92,7 @@ Paths_clickhouse_migrate_test_support RepairBoundaries RepairIntegration+ ReplicatedIntegration ReviewRemediation RunnerIntegration SurvivingWork@@ -98,8 +100,8 @@ autogen-modules: Paths_clickhouse_migrate_test_support ghc-options: -threaded build-depends:- clickhouse-migrate-cli ==0.1.*,- clickhouse-migrate-embed ==0.1.*,+ clickhouse-migrate-cli ==0.2.*,+ clickhouse-migrate-embed ==0.2.*, clickhouse-migrate-test-support, containers >=0.7 && <0.9,
src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs view
@@ -3,6 +3,7 @@ ( Fixture, Server (..), withFixture,+ withFixtureMode, fixtureRpc, fixtureEndpoint, fixtureRoot,@@ -38,13 +39,16 @@ field _ _ = fail "Invalid fixture response object" withFixture :: Text -> (Fixture -> IO a) -> IO a-withFixture entry action = mask $ \restore -> do+withFixture = withFixtureMode []++withFixtureMode :: [String] -> Text -> (Fixture -> IO a) -> IO a+withFixtureMode options entry action = mask $ \restore -> do script <- getDataFileName "test/scripts/native_fixture.py" matrix <- getDataFileName "test/data/clickhouse-matrix.json" (logPath, logHandle) <- openTempFile "/tmp" "clickhouse-migrate-fixture.log" pipes <- createProcess- ( (proc "python3" [script, "--matrix", matrix, "--entry", Text.unpack entry])+ ( (proc "python3" ([script, "--matrix", matrix, "--entry", Text.unpack entry] ++ options)) { std_in = CreatePipe, std_out = CreatePipe, std_err = UseHandle logHandle@@ -54,7 +58,7 @@ case pipes of (Just input, Just output, _, process) -> do ready <- try @SomeException $ do- line <- timeout 40000000 (BS.hGetLine output)+ line <- timeout (if null options then 40000000 else 90000000) (BS.hGetLine output) value <- maybe (fail "Fixture startup deadline expired") (either (const (fail "Fixture startup failed")) pure . Aeson.eitherDecodeStrict') line endpoint <- field "endpoint" value root <- field "root" value@@ -75,7 +79,7 @@ result <- try @SomeException (restore (action fixture)) cleanup <- try @SomeException $ do _ <- fixtureRpc fixture (object ["operation" .= ("stop" :: Text), "success" .= either (const False) (const True) result])- stopped <- timeout 20000000 (waitForProcess process)+ stopped <- timeout 60000000 (waitForProcess process) case stopped of Nothing -> terminateProcess process >> void (timeout 20000000 (waitForProcess process)) >> fail "Fixture cleanup deadline expired" Just _ -> pure ()@@ -99,7 +103,7 @@ Lazy.hPutStrLn input (Aeson.encode request) -- ByteString writes to a Handle still use its buffering policy. hFlush input- let deadline = case request of Object fields | KM.lookup "operation" fields == Just (String "restart") -> 40000000; _ -> 15000000+ let deadline = case request of Object fields | KM.lookup "operation" fields == Just (String "restart") -> 40000000; Object fields | KM.lookup "operation" fields == Just (String "start-replica") -> 70000000; Object fields | KM.lookup "operation" fields == Just (String "stop") -> 60000000; _ -> 15000000 line <- timeout deadline (BS.hGetLine output) reply <- maybe (fail ("Fixture RPC deadline expired; evidence: " ++ root)) (either (const (fail "Invalid fixture protocol")) pure . Aeson.eitherDecodeStrict') line case reply of
src/ClickHouseMigrate/TestSupport/Server.hs view
@@ -2,6 +2,10 @@ module ClickHouseMigrate.TestSupport.Server ( Server, withCandidateServer,+ withReplicatedCandidate,+ queryReplica,+ stopReplica,+ startReplica, serverEndpoint, serverLogDirectory, serverVersion,@@ -32,3 +36,17 @@ -- | SIGKILL/restart the owned candidate using the same temporary data root. restartCandidateServer :: Server -> IO () restartCandidateServer (Server fixture) = void $ fixtureRpc fixture (object ["operation" .= ("restart" :: Text)])++-- | Keeper and two owned Linux replicas behind a TCP round-robin endpoint.+withReplicatedCandidate :: Text -> (Server -> IO a) -> IO a+withReplicatedCandidate entry action = withFixtureMode ["--replicated"] entry (action . Server)++-- | Complete HTTP observation, including body/status, for independent assertions.+queryReplica :: Server -> Int -> Text -> IO Value+queryReplica (Server fixture) replica sql = fixtureRpc fixture (object ["operation" .= ("query-replica" :: Text), "replica" .= replica, "sql" .= sql])++stopReplica :: Server -> Int -> IO ()+stopReplica (Server fixture) replica = void $ fixtureRpc fixture (object ["operation" .= ("stop-replica" :: Text), "replica" .= replica])++startReplica :: Server -> Int -> IO ()+startReplica (Server fixture) replica = void $ fixtureRpc fixture (object ["operation" .= ("start-replica" :: Text), "replica" .= replica])
test/Main.hs view
@@ -26,6 +26,7 @@ import LedgerIntegration (ledgerFaultCases) import RepairBoundaries (repairBoundaryCases) import RepairIntegration (repairEarlyCases)+import ReplicatedIntegration (replicatedCases) import ReviewRemediation (remediationCases) import RunnerIntegration (runnerEarlyCases, runnerValidationCases) import SurvivingWork (runnerChild, survivingWorkCases)@@ -56,26 +57,33 @@ integrationMain :: [String] -> IO () integrationMain args = do- driver <- if args `elem` [["exclusive"], ["runner"], ["repair"], ["boundaries"], ["survivors"], ["adoption"], ["consumer"], ["cli"], ["example"], ["remediation"]] then pure "" else getEnv "CLICKHOUSE_MIGRATE_TEST_DRIVER"- finished <- timeout (if args == ["consumer"] then 600000000 else 180000000) $ withCandidateServer "lts-native" $ \server -> do+ let name = case args of [] -> "transport"; [group] -> group; _ -> ""+ (needsDriver, run) <- maybe (fail "Unknown integration group") pure (lookup name integrationGroups)+ driver <- if needsDriver then getEnv "CLICKHOUSE_MIGRATE_TEST_DRIVER" else pure ""+ candidate <- maybe "lts-native" Text.pack <$> lookupEnv "CLICKHOUSE_MIGRATE_CANDIDATE"+ finished <- timeout (if name `elem` ["consumer", "replicated"] then 600000000 else 180000000) $ (if name == "replicated" then withReplicatedCandidate else withCandidateServer) candidate $ \server -> do putStrLn ("Candidate ClickHouse " ++ Text.unpack (serverVersion server))- case args of- [] -> transportCases driver server- ["transport"] -> transportCases driver server- ["completion"] -> completionCases driver server- ["ledger"] -> ledgerCases driver server- ["consumer"] -> consumerFixtureCases server- ["cli"] -> cliCases server- ["example"] -> exampleCases server- ["adoption"] -> adoptionEarlyCases server >> adoptionFaultCases server- ["remediation"] -> remediationCases server- ["runner"] -> runnerEarlyCases server >> runnerValidationCases server- ["repair"] -> repairEarlyCases server >> repairBoundaryCases server- ["survivors"] -> survivingWorkCases server- ["boundaries"] -> failureBoundaryCases server- ["exclusive"] -> exclusiveCases server- _ -> fail "Unknown integration group"+ run driver server case finished of Nothing -> fail "Integration watchdog expired"; Just () -> pure ()++-- Every group and its private-driver requirement is declared once here.+integrationGroups :: [(String, (Bool, FilePath -> Server -> IO ()))]+integrationGroups =+ [ ("replicated", (False, const replicatedCases)),+ ("transport", (True, transportCases)),+ ("completion", (True, completionCases)),+ ("ledger", (True, ledgerCases)),+ ("consumer", (False, const consumerFixtureCases)),+ ("cli", (False, const cliCases)),+ ("example", (False, const exampleCases)),+ ("adoption", (False, \_ server -> adoptionEarlyCases server >> adoptionFaultCases server)),+ ("remediation", (False, const remediationCases)),+ ("runner", (False, \_ server -> runnerEarlyCases server >> runnerValidationCases server)),+ ("repair", (False, \_ server -> repairEarlyCases server >> repairBoundaryCases server)),+ ("survivors", (False, const survivingWorkCases)),+ ("boundaries", (False, const failureBoundaryCases)),+ ("exclusive", (False, const exclusiveCases))+ ] runDriver :: FilePath -> Server -> Text -> String -> String -> ByteString -> Int -> Int -> IO Value runDriver = runDriverExpected Nothing
+ test/ReplicatedIntegration.hs view
@@ -0,0 +1,181 @@+module ReplicatedIntegration (replicatedCases) where++import ClickHouseMigrate+import ClickHouseMigrate.Event qualified as E+import ClickHouseMigrate.Inspection qualified as I+import ClickHouseMigrate.Ledger qualified as L+import ClickHouseMigrate.Repair qualified as Repair+import ClickHouseMigrate.Report qualified as R+import ClickHouseMigrate.Runner qualified as Run+import ClickHouseMigrate.TestSupport.Server+import Control.Monad (forM, forM_, void, when)+import Data.Aeson (Value, object, (.=))+import Data.Aeson qualified as JSON+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as Lazy+import Data.List.NonEmpty (NonEmpty (..))+import Data.List.NonEmpty qualified as NE+import Data.Set qualified as Set+import Data.Text qualified as T+import Data.Text.Encoding qualified as Encoding+import RunnerIntegration (assert, field, provider, right)+import System.Environment (getEnv, getEnvironment)+import System.Exit (ExitCode (..))+import System.FilePath ((</>))+import System.Process (CreateProcess (..), proc, readCreateProcessWithExitCode)+import System.Timeout (timeout)++rowsAt :: Server -> Int -> T.Text -> IO [Value]+rowsAt server replica sql = do+ result <- queryReplica server replica sql+ httpStatus <- field "status" result :: IO Int+ body <- field "body" result+ assert ("Independent read failed: " ++ T.unpack body) (httpStatus == 200)+ mapM (right . JSON.eitherDecodeStrict' . Encoding.encodeUtf8) (filter (not . T.null) (T.lines body))++replicatedCases :: Server -> IO ()+replicatedCases server = do+ replicas <- forM [0, 1] $ \replica -> do+ rows <- rowsAt server replica "SELECT toString(serverUUID()) AS id"+ case rows of [row] -> field "id" row >>= right . serverId; _ -> fail "Missing replica identity"+ ids <- case replicas of a : b : [] -> pure (a :| [b]); _ -> fail "Missing replicas"+ cluster <- right (clusterName "fixture")+ endpoint <- right (directEndpoint (serverEndpoint server))+ credentials <- right (basicCredentials "app" "test-password")+ access <- provider+ audit <- right (L.auditContext "replicated fixture" "owned fixture; predecessor requests drained" ["fixture/controlled-drain"])+ let configAt key namespace binding ep = right (replicatedTargetConfig key ep cluster ids namespace binding credentials defaultHttpLimits)+ configure name binding = do+ key <- right (targetKey ("replicated/" <> name))+ namespace <- right (ledgerNamespace ("fixture_" <> name))+ configAt key namespace binding endpoint+ createNamespace name = void (queryServer server ("CREATE DATABASE fixture_" <> name <> " ON CLUSTER fixture ENGINE=Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')"))+ context config sink = Run.mutationContext config access sink E.redactedDiagnostics+ observe config plan = I.status I.Observe config plan >>= right+ -- An actual application-mounted executable, through the alternating endpoint.+ example <- getEnv "CLICKHOUSE_MIGRATE_EXAMPLE"+ inherited <- getEnvironment+ exampleLedger <- newLedgerId+ let exampleKey = "replicated/example"+ configured = [("CHM_CLUSTER", "fixture"), ("CHM_REPLICA_IDS", T.unpack (T.intercalate "," (map serverIdText replicas))), ("CHM_TARGET", exampleKey), ("CHM_ENDPOINT", T.unpack (serverEndpoint server)), ("CHM_USER", "app"), ("CHM_PASSWORD", "test-password"), ("CHM_LEDGER_DATABASE", "fixture_ledger"), ("CHM_LOCK_ROOT", serverLogDirectory server </> "example-lock")]+ run extra args = do+ let values = extra ++ configured+ env = values ++ filter (\(name, _) -> name `notElem` map fst values) inherited+ (exit, output, errors) <- readCreateProcessWithExitCode ((proc example ("--replicated-example" : args)) {env = Just env}) ""+ assert ("Replicated mounted example failed: " ++ output ++ errors) (exit == ExitSuccess)+ value <- right (JSON.eitherDecodeStrict' (Encoding.encodeUtf8 (T.pack output)) :: Either String Value)+ pure value+ upArgs = ["up", "--json", "--expect-target", exampleKey, "--actor", "fixture", "--reason", "owned predecessor drained", "--evidence", "fixture/drain", "--quiescent"]+ bound = [("CHM_LEDGER_ID", T.unpack (ledgerIdText exampleLedger))]+ offline <- run [] ["plan", "--json"]+ components <- field "result" offline >>= field "components" :: IO [Value]+ authored <- concat <$> mapM (field "migrations") components :: IO [Value]+ expectedDigests <- mapM (\migration -> field "checksums" migration >>= field "sha256-v2") authored :: IO [T.Text]+ pending <- run [] ["status", "--json"]+ summaries <- field "result" pending >>= field "report" >>= field "migrations" :: IO [Value]+ pendingDigests <- mapM (\migration -> field "definition" migration >>= field "checksum") summaries :: IO [T.Text]+ assert "Empty replicated ledger status used v1 pending checksums" (pendingDigests == expectedDigests)+ void (run [] (upArgs ++ ["--initialize-ledger", T.unpack (ledgerIdText exampleLedger)]))+ before <- rowsAt server 0 "SELECT event FROM fixture_ledger.events ORDER BY sequence"+ void (run bound upArgs)+ after <- rowsAt server 1 "SELECT event FROM fixture_ledger.events ORDER BY sequence"+ assert "Repeated up appended history or replicas diverged" (before == after)+ statusJson <- run bound ["status", "--json"]+ format <- field "result" statusJson >>= field "report" >>= field "observation" >>= field "checksumFormat" :: IO T.Text+ assert "Replicated status is not v2" (format == "sha256-v2")+ void (run bound ["history", "--json"])+ void (run bound ["verify", "--require-current", "--json"])+ forM_ ["chm_replicated_catalog", "chm_replicated_events"] $ \table -> do+ a <- rowsAt server 0 ("SELECT * FROM fixture_app." <> table)+ b <- rowsAt server 1 ("SELECT * FROM fixture_app." <> table)+ assert "Acknowledged application data differs between replicas" (a == b && not (null a))+ forM_ [0, 1] $ \replica -> do+ stopReplica server replica+ startReplica server replica+ durable <- rowsAt server replica "SELECT event FROM fixture_ledger.events ORDER BY sequence"+ assert "Acknowledged ledger lost after restart" (durable == before)+ putStrLn "replicated: mounted up/status/history/verify; v2; no-op reapply; identical data/history; both process restarts"+ -- Missing, Atomic, malformed and foreign-format ledgers fail closed.+ minimal <- makePlan [("one", "CREATE TABLE fixture_app.shape_probe (x UInt8) ENGINE=ReplicatedMergeTree ORDER BY x")]+ missing <- configure "missing" FirstUse+ absent <- I.history I.Observe missing+ assert "Missing replicated ledger database accepted" (case absent of Left (R.InspectionFailure (R.LedgerFailed L.LedgerDatabaseMissing) _ _) -> True; _ -> False)+ void (queryServer server "CREATE DATABASE fixture_atomic ON CLUSTER fixture ENGINE=Atomic")+ atomic <- configure "atomic" FirstUse+ rejected <- I.history I.Observe atomic+ assert "Atomic replicated ledger accepted" (case rejected of Left (R.InspectionFailure (R.LedgerFailed L.InvalidLedgerSchema) _ _) -> True; _ -> False)+ createNamespace "badshape"+ void (queryServer server "CREATE TABLE fixture_badshape.events (sequence UInt64, event_id UUID, ledger_id UUID, event String) ENGINE=Memory")+ badshape <- configure "badshape" FirstUse+ malformed <- I.history I.Observe badshape+ assert "Wrong ledger engine accepted" (case malformed of Left (R.InspectionFailure (R.LedgerFailed L.InvalidLedgerSchema) _ _) -> True; _ -> False)+ createNamespace "foreign"+ void (queryServer server "CREATE TABLE fixture_foreign.events (sequence UInt64, event_id UUID, ledger_id UUID, event String) ENGINE=ReplicatedMergeTree ORDER BY sequence SETTINGS fsync_after_insert=1, fsync_part_directory=1, storage_policy='default'")+ originalInitialization <- case before of row : _ -> field "event" row; _ -> fail "Missing initialization for foreign-format test"+ decodedInitialization <- right (JSON.eitherDecodeStrict' (Encoding.encodeUtf8 originalInitialization) :: Either String Value)+ eventId <- field "eventId" decodedInitialization :: IO T.Text+ oldPolicy <- field "body" decodedInitialization >>= field "policy"+ let foreignEvent = T.replace "replicated/example" "replicated/foreign" (T.replace oldPolicy (Encoding.decodeUtf8 completionPolicy) (T.replace "sha256-v2" "sha256-v1" originalInitialization))+ foreignRow = object ["sequence" .= (1 :: Int), "event_id" .= eventId, "ledger_id" .= ledgerIdText exampleLedger, "event" .= foreignEvent]+ void (queryServer server ("INSERT INTO fixture_foreign.events FORMAT JSONEachRow\n" <> Encoding.decodeUtf8 (Lazy.toStrict (JSON.encode foreignRow))))+ forM_ [0, 1] $ \replica -> void (rowsAt server replica "SYSTEM SYNC REPLICA fixture_foreign.events")+ foreignConfig <- configure "foreign" FirstUse+ foreignFormat <- I.history I.Observe foreignConfig+ assert "Replicated target accepted a v1 ledger" (case foreignFormat of Left (R.InspectionFailure (R.LedgerFailed L.UnsupportedLedgerVersion) _ _) -> True; _ -> False)+ identityResult <- queryReplica server 0 "SELECT 1"+ directEndpointText <- field "endpoint" identityResult+ directEndpointValue <- right (directEndpoint directEndpointText)+ key <- right (targetKey "replicated/membership")+ namespace <- right (ledgerNamespace "fixture_ledger")+ wrongCount <- right (replicatedTargetConfig key directEndpointValue cluster (NE.head ids :| []) namespace FirstUse credentials defaultHttpLimits)+ wrong <- I.history I.Observe wrongCount+ assert "Wrong membership count accepted" (case wrong of Left (R.InspectionFailure (R.TargetFailed (L.TransportError L.ReplicaSetMismatch _ _)) _ _) -> True; _ -> False)+ -- Database creation is rejected before even initializing the pending ledger.+ createNamespace "database_step"+ databaseConfig <- configure "database_step" FirstUse+ databasePlan <- makePlan [("one", "/* before header */ CREATE DATABASE forbidden_fixture_database")]+ databaseLedger <- newLedgerId+ databaseResult <- Run.up (context databaseConfig E.quietEvents) (Run.upOptions audit (Just (L.authorizeInitialization (configuredTargetKey databaseConfig) databaseLedger audit))) databasePlan+ assert "CREATE DATABASE reached replicated runner" (case databaseResult of Left (R.RunFailure (R.DatabaseCreationOnReplicatedTarget _) _ _) -> True; _ -> False)+ objects <- rowsAt server 0 "SELECT name FROM system.tables WHERE database='fixture_database_step'"+ assert "Preflight initialized a ledger before rejecting database creation" (null objects)+ -- Plain MergeTree reaches the server and is classified with Code 56.+ createNamespace "plain"+ plainConfig <- configure "plain" FirstUse+ plainLedger <- newLedgerId+ plainPlan <- makePlan [("one", "CREATE TABLE fixture_app.plain (x UInt8) ENGINE=MergeTree ORDER BY x")]+ plain <- Run.up (context plainConfig E.quietEvents) (Run.upOptions audit (Just (L.authorizeInitialization (configuredTargetKey plainConfig) plainLedger audit))) plainPlan+ assert "Code 56 was not an observed SQL failure" (case plain of Left (R.RunFailure (R.SqlFailed _ _ L.ObservedFailure (L.TransportError _ _ (Just 56))) _ _) -> True; _ -> False)+ putStrLn "replicated: missing/Atomic/wrong ledger shapes, membership mismatch and pending CREATE DATABASE rejected; plain MergeTree observed Code 56"+ -- Stop the other replica only after intent is acknowledged, so the actual DDL+ -- failure and the dependent ledger uncertainty are both observable.+ createNamespace "failure"+ first <- configure "failure" FirstUse+ firstDirect <- configAt (configuredTargetKey first) (either (error . show) id (ledgerNamespace "fixture_failure")) FirstUse directEndpointValue+ ledger <- newLedgerId+ let sink (E.RunEvent phase _ _ _ _ _ _) = when (phase == E.BeforeSql) (stopReplica server 1)+ failed <- timeout 65000000 (Run.up (context firstDirect sink) (Run.upOptions audit (Just (L.authorizeInitialization (configuredTargetKey first) ledger audit))) minimal)+ assert "Missing replica did not fail within bounded request waits" (case failed of Just (Left (R.RunFailure (R.SqlFailed _ _ L.ObservedFailure _) _ _)) -> True; _ -> False)+ startReplica server 1+ void (rowsAt server 1 "SYSTEM SYNC DATABASE REPLICA fixture_app")+ forM_ [0, 1] $ \replica -> void (rowsAt server replica "SYSTEM SYNC REPLICA fixture_failure.events")+ boundConfig <- configAt (configuredTargetKey first) (either (error . show) id (ledgerNamespace "fixture_failure")) (ExistingLedger ledger) directEndpointValue+ R.InspectionReport (R.HistoryReport _ _ _ (Just position) _ _ _ _) [R.MigrationSummary _ _ (Just attempt) [R.StepSummary _ _ state (Just selected) _]] _ <- observe boundConfig minimal+ assert "Failed DDL was recorded as succeeded" (state `elem` [R.StepUncertain, R.StepObservedFailure])+ quiet <- right (L.attestQuiescence (configuredTargetKey first) audit)+ evidence <- right (Repair.recoveryEvidence audit quiet [])+ accept <- right (Repair.acceptStepRequest position attempt selected evidence)+ void (Repair.repair (context boundConfig E.quietEvents) minimal accept >>= right)+ R.InspectionReport (R.HistoryReport _ _ _ (Just next) _ _ _ _) [R.MigrationSummary _ _ (Just nextAttempt) _] _ <- observe boundConfig minimal+ finalize <- right (Repair.finalizeRequest next nextAttempt evidence)+ void (Repair.repair (context boundConfig E.quietEvents) minimal finalize >>= right)+ void (I.verify I.Observe boundConfig (I.verifyOptions True) minimal >>= right)+ putStrLn "replicated: stopped replica produced server failure and uncertain ledger append; restart, explicit accept/finalize and verify succeeded"++makePlan :: [(T.Text, BS.ByteString)] -> IO MigrationPlan+makePlan sources = do+ migrations <- forM sources $ \(name, sql) -> do+ step <- right (sqlStep "execute" sql)+ right (sqlMigration name (step :| []))+ component <- right (migrationComponentFromList "replicated" Set.empty migrations)+ right (migrationPlan (component :| []))
test/data/clickhouse-matrix.json view
@@ -1,7 +1,7 @@ { "matrixVersion": 1,- "checkedAt": "2026-10-16",- "policy": "Only exact qualified patches; no prereleases, Cloud, proxies, distributed or replicated execution",+ "checkedAt": "2026-11-07",+ "policy": "Only exact qualified patches and declared topology groups; no prereleases or managed-cloud certification", "entries": [ { "id": "lts-native",@@ -23,7 +23,135 @@ "Nix public suite, private driver and example" ], "scope": "Native protocol qualification; no production OS, Linux, power-loss or uptime certification",- "startupCaveat": "A separate attempt exited SIGSEGV before readiness; cause unknown; preserved in docs/release-evidence.md"+ "startupCaveat": "A separate attempt exited SIGSEGV before readiness; cause unknown; preserved in docs/release-evidence.md",+ "requalification": {+ "date": "2026-11-02",+ "requiredGroups": 13,+ "artifactVariants": [+ "Nix suite, private driver and example",+ "seven extracted sibling source distributions"+ ]+ }+ },+ "groups": [+ "transport",+ "completion",+ "ledger",+ "exclusive",+ "runner",+ "repair",+ "boundaries",+ "survivors",+ "adoption",+ "consumer",+ "cli",+ "example",+ "remediation"+ ]+ },+ {+ "id": "altinity-linux-arm64",+ "version": "26.3.60.10001.altinitystable",+ "platform": "aarch64-linux",+ "architecture": "arm64",+ "image": "docker.io/altinity/clickhouse-server@sha256:3bafa18ced8daf9b81de70927b40c13b06545363abe3ad7f20ae77848f500c5a",+ "status": "native-qualified",+ "scope": "Owned Linux arm64 single-server and single-shard replicated fixtures; no managed-cloud certification",+ "tag": "v26.3.60.10001.altinitystable",+ "commit": "f01425ddbe11657d479571ba5364b7bb0e9bc2e0",+ "releaseDate": "2026-10-29",+ "qualification": {+ "date": "2026-11-02",+ "compiler": "ghc-9.12.4",+ "clientHost": "macOS 26.7 arm64",+ "requiredGroups": 13,+ "artifactVariants": [+ "Nix suite, private driver and example",+ "seven extracted sibling source distributions"+ ],+ "scope": "Owned single local-server Linux arm64 container; no managed-cloud, Linux client or power-loss claim",+ "startupCaveat": "Docker-host port collision fixed by allocating and retaining ports on the Docker host. Final 26-case Nix and 26-case source variants pass; no startup retries."+ },+ "groups": [+ "transport",+ "completion",+ "ledger",+ "exclusive",+ "runner",+ "repair",+ "boundaries",+ "survivors",+ "adoption",+ "consumer",+ "cli",+ "example",+ "remediation",+ "replicated"+ ],+ "replicatedQualification": {+ "date": "2026-11-07",+ "cohort": "0.1.1.0",+ "shards": 1,+ "replicas": 2,+ "group": "replicated",+ "variants": [+ "Cabal full 41-case matrix",+ "Nix installed replicated group for each Altinity image"+ ],+ "evidence": "docs/release-evidence/2026-11-07-replicated.md"+ }+ },+ {+ "id": "altinity-linux-arm64-10002",+ "version": "26.3.60.10002.altinitystable",+ "platform": "aarch64-linux",+ "architecture": "arm64",+ "image": "docker.io/altinity/clickhouse-server@sha256:7e72405bf85df6d163205ee2e56251ebc35daf6dc71142f58bdee7a461690ffb",+ "status": "native-qualified",+ "scope": "Owned Linux arm64 single-server and single-shard replicated fixtures; no managed-cloud certification",+ "tag": "v26.3.60.10002-1.altinitystable",+ "commit": "4d1d8cad81573563e3abece05c39fb4903fce73f",+ "releaseDate": "2026-11-05",+ "qualification": {+ "date": "2026-11-07",+ "compiler": "ghc-9.12.4",+ "clientHost": "macOS arm64",+ "requiredGroups": 13,+ "artifactVariants": [+ "Nix suite, private driver and example",+ "seven extracted sibling source distributions"+ ],+ "candidateGroupCasesPerVariant": 39,+ "scope": "Owned single-server Linux arm64 artifact; replication and managed-cloud acceptance are separate"+ },+ "indexDigest": "sha256:53e81c410a22ab052a4d967adc2886b547c91e67370e2d02f5c2ea1177d4d3aa",+ "groups": [+ "transport",+ "completion",+ "ledger",+ "exclusive",+ "runner",+ "repair",+ "boundaries",+ "survivors",+ "adoption",+ "consumer",+ "cli",+ "example",+ "remediation",+ "replicated"+ ],+ "replicatedQualification": {+ "date": "2026-11-07",+ "cohort": "0.1.1.0",+ "shards": 1,+ "replicas": 2,+ "group": "replicated",+ "variants": [+ "Cabal full 41-case matrix",+ "Nix installed replicated group for each Altinity image"+ ],+ "evidence": "docs/release-evidence/2026-11-07-replicated.md" } } ],
test/scripts/native_fixture.py view
@@ -19,6 +19,7 @@ import time import urllib.request import urllib.parse+import uuid from xml.sax.saxutils import escape PASSWORD = 'test-password'@@ -58,12 +59,27 @@ class NativeServer:- def __init__(self, entry):+ def __init__(self, entry, deployment=None): self.entry = entry self.root = Path(tempfile.mkdtemp(prefix='clickhouse-migrate-server-', dir='/tmp')) self.port = port() self.tcp_port = port()- self.binary = binary_for(entry)+ self.container = None+ self.docker = None+ if 'image' in entry:+ context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')+ if not context:+ raise RuntimeError('OCI candidate requires an explicit isolated Docker context')+ self.docker = ['docker', '--context', context]+ endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']+ if not endpoint.startswith('unix://'):+ raise RuntimeError('Owned OCI fixture requires a local Unix Docker socket')+ image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]+ if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':+ raise RuntimeError('OCI candidate platform mismatch; pre-pull the exact matrix digest')+ self.binary = None+ else:+ self.binary = binary_for(entry) self.process = None self.log = (self.root / 'console.log').open('w') self.proxies = []@@ -76,12 +92,19 @@ <profile>default</profile><quota>default</quota><access_management>1</access_management></default></users> <quotas><default><interval><duration>3600</duration><queries>0</queries><errors>0</errors> <result_rows>0</result_rows><read_rows>0</read_rows><execution_time>0</execution_time></interval></default></quotas></clickhouse>''')- root = escape(str(self.root))+ root = '/fixture' if self.docker else escape(str(self.root))+ if self.docker:+ users.write_text(users.read_text().replace('<ip>127.0.0.1</ip>', '<ip>::/0</ip>'))+ users_path = '/fixture/users.xml' if self.docker else str(users)+ listener = '0.0.0.0' if self.docker else '127.0.0.1'+ http_port = 8123 if self.docker else self.port+ tcp_port = 9000 if self.docker else self.tcp_port (self.root / 'config.xml').write_text(f'''<clickhouse> <logger><level>warning</level><log>{root}/server.log</log><errorlog>{root}/error.log</errorlog></logger> <path>{root}/data/</path><tmp_path>{root}/tmp/</tmp_path><user_files_path>{root}/user_files/</user_files_path>-<format_schema_path>{root}/format_schemas/</format_schema_path><users_config>{root}/users.xml</users_config>-<listen_host>127.0.0.1</listen_host><http_port>{self.port}</http_port><tcp_port>{self.tcp_port}</tcp_port>+<access_control_path>{root}/access/</access_control_path>+<format_schema_path>{root}/format_schemas/</format_schema_path><users_config>{users_path}</users_config>+<listen_host>{listener}</listen_host><http_port>{http_port}</http_port><tcp_port>{tcp_port}</tcp_port> <max_connections>64</max_connections><max_thread_pool_size>128</max_thread_pool_size> <background_schedule_pool_size>8</background_schedule_pool_size><background_buffer_flush_schedule_pool_size>2</background_buffer_flush_schedule_pool_size> <background_message_broker_schedule_pool_size>2</background_message_broker_schedule_pool_size>@@ -90,8 +113,28 @@ <background_fetches_pool_size>2</background_fetches_pool_size><background_common_pool_size>2</background_common_pool_size> <mark_cache_size>0</mark_cache_size><uncompressed_cache_size>0</uncompressed_cache_size> </clickhouse>''')+ if deployment:+ extra = deployment['config']+ config = self.root / 'config.xml'+ config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>')) try:- self.process = subprocess.Popen([str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')], stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)+ if self.docker:+ name = deployment['name'] if deployment else 'clickhouse-migrate-' + uuid.uuid4().hex+ subprocess.run(self.docker + ['create', '--name', name, '--label', 'clickhouse-migrate.fixture=owned',+ *(['--network', deployment['network']] if deployment else []),+ '--user', str(os.getuid()), '--memory', '2g', '--cpus', '2',+ '-p', '127.0.0.1::8123', '-p', '127.0.0.1::9000',+ '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'],+ check=True, capture_output=True, timeout=20)+ self.container = name+ subprocess.run(self.docker + ['cp', str(self.root) + '/.', name + ':/fixture'], check=True, capture_output=True, timeout=20)+ # Docker allocates ports on its own host, not the client OS.+ # Keep the namespace alive while restarting only ClickHouse so+ # these exact bindings remain reserved for the fixture lifetime.+ subprocess.run(self.docker + ['start', name], check=True, capture_output=True, timeout=20)+ self.port = int(subprocess.check_output(self.docker + ['port', name, '8123/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])+ self.tcp_port = int(subprocess.check_output(self.docker + ['port', name, '9000/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])+ self.start() deadline = time.monotonic() + 30 while time.monotonic() < deadline: if self.process.poll() is not None:@@ -107,7 +150,7 @@ raise RuntimeError('Owned server readiness deadline expired') identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0] version = identity['version']- if identity['server_id'] != (self.root / 'data/uuid').read_text().strip():+ if identity['server_id'] != self.retained_uuid(): raise RuntimeError('Endpoint is not the owned server data directory') if version != entry['version']: raise RuntimeError('Candidate server version mismatch')@@ -125,19 +168,41 @@ with urllib.request.urlopen(request, timeout=10) as response: return [json.loads(line) for line in response.read().splitlines() if line.strip()] + def retained_uuid(self):+ if self.docker:+ return subprocess.check_output(self.docker + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip()+ return (self.root / 'data/uuid').read_text().strip()++ def start(self):+ command = self.docker + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.docker else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')]+ self.process = subprocess.Popen(command, stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)++ def stop(self, sig):+ if self.process is not None and self.process.poll() is None:+ if self.docker:+ subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10)+ else:+ os.killpg(self.process.pid, sig)+ try:+ self.process.wait(timeout=10)+ except subprocess.TimeoutExpired:+ if self.docker:+ subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10)+ else:+ os.killpg(self.process.pid, signal.SIGKILL)+ self.process.wait(timeout=5)+ def restart(self): """Kill/restart only this owned process, preserving its data/config/UUID."""- if self.process is not None and self.process.poll() is None:- os.killpg(self.process.pid, signal.SIGKILL)- self.process.wait(timeout=5)- self.process = subprocess.Popen([str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')], stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)+ self.stop(signal.SIGKILL)+ self.start() deadline = time.monotonic() + 30 while time.monotonic() < deadline: if self.process.poll() is not None: raise RuntimeError('Owned restarted server exited') try: identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]- if identity['version'] == self.entry['version'] and identity['server_id'] == (self.root / 'data/uuid').read_text().strip():+ if identity['version'] == self.entry['version'] and identity['server_id'] == self.retained_uuid(): return except OSError: pass@@ -147,14 +212,18 @@ def close(self, success): for proxy in self.proxies: proxy.close()- if self.process is not None and self.process.poll() is None:- os.killpg(self.process.pid, signal.SIGTERM)- try:- self.process.wait(timeout=10)- except subprocess.TimeoutExpired:- os.killpg(self.process.pid, signal.SIGKILL)- self.process.wait(timeout=5)- self.log.close()+ try:+ self.stop(signal.SIGTERM)+ finally:+ if self.container:+ # Copy only this fixture's private data/logs before removing its+ # recorded container. No ambient container or host volume is used.+ if not success:+ evidence = self.root / 'container-evidence'+ evidence.mkdir()+ subprocess.run(self.docker + ['cp', self.container + ':/fixture/.', str(evidence)], capture_output=True, timeout=30, check=True)+ subprocess.run(self.docker + ['rm', '--force', self.container], capture_output=True, timeout=15, check=True)+ self.log.close() if success: shutil.rmtree(self.root) @@ -389,13 +458,17 @@ (self.target.root / f'proxy-{self.http.server_port}-events.json').write_text(json.dumps(evidence, indent=2) + '\n') -def serve(matrix, entry_id):- entries = json.loads(Path(matrix).read_text())['entries']- entry = next(value for value in entries if value['id'] == entry_id)- server = NativeServer(entry)+def serve(matrix, entry_id, replicated=False): def stop_signal(*_): raise SystemExit(1) signal.signal(signal.SIGTERM, stop_signal)+ entries = json.loads(Path(matrix).read_text())['entries']+ entry = next(value for value in entries if value['id'] == entry_id)+ if replicated:+ from replicated_fixture import ReplicatedServer+ server = ReplicatedServer(entry)+ else:+ server = NativeServer(entry) print(json.dumps({'endpoint': server.endpoint, 'root': str(server.root), 'version': entry['version']}), flush=True) success = False try:@@ -403,7 +476,15 @@ request = json.loads(line) operation = request['operation'] try:- if operation == 'restart':+ if operation == 'query-replica':+ result = server.request(request['replica'], request['sql'], request.get('settings'), request.get('user', 'default'))+ elif operation == 'stop-replica':+ server.stop_replica(request['replica'])+ result = None+ elif operation == 'start-replica':+ server.start_replica(request['replica'])+ result = None+ elif operation == 'restart': server.restart() result = None elif operation == 'query':@@ -461,5 +542,6 @@ parser = argparse.ArgumentParser() parser.add_argument('--matrix', required=True) parser.add_argument('--entry', default='lts-native')+ parser.add_argument('--replicated', action='store_true') options = parser.parse_args()- serve(options.matrix, options.entry)+ serve(options.matrix, options.entry, options.replicated)
+ test/scripts/replicated_fixture.py view
@@ -0,0 +1,195 @@+#!/usr/bin/env python3+"""Owned Keeper plus two replicas; no shared server or Docker network."""+import base64+import http.client+import json+import os+from pathlib import Path+import select+import shutil+import signal+import socket+import socketserver+import subprocess+import tempfile+import threading+import time+import urllib.parse+import uuid++from native_fixture import NativeServer, PASSWORD+++class RoundRobin:+ def __init__(self, replicas):+ self.replicas = replicas+ self.next = 0+ self.lock = threading.Lock()+ owner = self++ class Handler(socketserver.BaseRequestHandler):+ def handle(self):+ with owner.lock:+ target = owner.replicas[owner.next % len(owner.replicas)]+ owner.next += 1+ try:+ with socket.create_connection(('127.0.0.1', target.port), timeout=2) as upstream:+ peers = [self.request, upstream]+ while True:+ ready, _, _ = select.select(peers, [], [], 1)+ if not ready:+ if owner.closed:+ return+ continue+ for source in ready:+ data = source.recv(65536)+ if not data:+ return+ (upstream if source is self.request else self.request).sendall(data)+ except OSError:+ return+ class Listener(socketserver.ThreadingTCPServer):+ daemon_threads = True+ allow_reuse_address = True+ self.closed = False+ self.server = Listener(('127.0.0.1', 0), Handler)+ self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)+ self.thread.start()+ self.endpoint = 'http://127.0.0.1:' + str(self.server.server_address[1])++ def close(self):+ self.closed = True+ self.server.shutdown()+ self.server.server_close()+ self.thread.join(timeout=2)+++class ReplicatedServer:+ def __init__(self, entry):+ if 'image' not in entry:+ raise RuntimeError('Replicated fixture requires an OCI candidate')+ self.entry = entry+ self.root = Path(tempfile.mkdtemp(prefix='clickhouse-migrate-replicated-', dir='/tmp'))+ context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')+ if not context:+ raise RuntimeError('Replicated fixture requires an explicit isolated Docker context')+ self.docker = ['docker', '--context', context]+ endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']+ if not endpoint.startswith('unix://'):+ raise RuntimeError('Owned replicated fixture requires a local Unix Docker socket')+ image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]+ if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':+ raise RuntimeError('Replicated OCI candidate platform mismatch; pre-pull the exact matrix digest')+ self.network = 'clickhouse-migrate-' + uuid.uuid4().hex+ self.keeper = None+ self.replicas = []+ self.proxy = None+ self.closed = False+ try:+ subprocess.run(self.docker + ['network', 'create', '--label', 'clickhouse-migrate.fixture=owned', self.network], check=True, capture_output=True)+ keeper_name = self.network + '-keeper'+ self.keeper = keeper_name+ keeper_config = self.root / 'keeper.xml'+ keeper_config.write_text('''<clickhouse><logger><level>warning</level><console>true</console></logger>+<listen_host>0.0.0.0</listen_host><keeper_server><tcp_port>9181</tcp_port><server_id>1</server_id>+<log_storage_path>/fixture/log</log_storage_path><snapshot_storage_path>/fixture/snapshots</snapshot_storage_path>+<coordination_settings><operation_timeout_ms>10000</operation_timeout_ms><session_timeout_ms>10000</session_timeout_ms></coordination_settings>+<raft_configuration><server><id>1</id><hostname>localhost</hostname><port>9234</port></server></raft_configuration>+</keeper_server></clickhouse>''')+ subprocess.run(self.docker + ['create', '--name', keeper_name, '--network', self.network, '--label', 'clickhouse-migrate.fixture=owned', '--user', str(os.getuid()), '-p', '127.0.0.1::9181', '--memory', '512m', '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'], check=True, capture_output=True)+ subprocess.run(self.docker + ['cp', str(self.root) + '/.', keeper_name + ':/fixture'], check=True, capture_output=True)+ subprocess.run(self.docker + ['start', keeper_name], check=True, capture_output=True)+ self.keeper_log = (self.root / 'keeper.log').open('w')+ self.keeper_process = subprocess.Popen(self.docker + ['exec', keeper_name, 'clickhouse', 'keeper', '--config-file=/fixture/keeper.xml'], stdout=self.keeper_log, stderr=subprocess.STDOUT)+ keeper_port = int(subprocess.check_output(self.docker + ['port', keeper_name, '9181/tcp'], text=True).strip().rsplit(':', 1)[1])+ deadline = time.monotonic() + 20+ while time.monotonic() < deadline:+ if self.keeper_process.poll() is not None:+ raise RuntimeError('Keeper exited before readiness')+ try:+ with socket.create_connection(('127.0.0.1', keeper_port), timeout=.2) as connection:+ connection.sendall(b'ruok')+ if connection.recv(4) == b'imok':+ break+ except OSError:+ pass+ time.sleep(.05)+ else:+ raise RuntimeError('Keeper readiness deadline expired')+ names = [self.network + '-r1', self.network + '-r2']+ cluster = '<remote_servers><fixture><shard>' + ''.join('<replica><host>' + name + '</host><port>9000</port><user>default</user><password>' + PASSWORD + '</password></replica>' for name in names) + '</shard></fixture></remote_servers>'+ for i, name in enumerate(names):+ extra = cluster + '<zookeeper><node><host>' + keeper_name + '</host><port>9181</port></node></zookeeper><macros><cluster>fixture</cluster><shard>1</shard><replica>r' + str(i + 1) + '</replica></macros><interserver_http_port>9009</interserver_http_port><interserver_http_host>' + name + '</interserver_http_host><distributed_ddl><path>/clickhouse/task_queue/ddl</path></distributed_ddl>'+ self.replicas.append(NativeServer(entry, dict(name=name, network=self.network, config=extra)))+ (self.root / 'replicas.json').write_text(json.dumps([dict(replica='r' + str(n + 1), root=str(replica.root), endpoint=replica.endpoint) for n, replica in enumerate(self.replicas)], indent=2) + '\n')+ # Administrative bootstrap exactly mirrors consumer infrastructure.+ for database in ['fixture_app', 'fixture_ledger']:+ result = self.request(0, "CREATE DATABASE " + database + " ON CLUSTER fixture ENGINE=Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')", user='default')+ if result['status'] != 200:+ raise RuntimeError('Bootstrap: ' + result['body'])+ for replica in self.replicas:+ replica.query("CREATE USER app IDENTIFIED WITH plaintext_password BY 'test-password' SETTINGS database_replicated_allow_only_replicated_engine = 1 CONST")+ replica.query('GRANT SELECT, INSERT, CREATE TABLE, CREATE VIEW, ALTER TABLE, DROP TABLE, DROP VIEW, TRUNCATE ON *.* TO app')+ self.proxy = RoundRobin(self.replicas)+ self.endpoint = self.proxy.endpoint+ except BaseException:+ self.close(False)+ raise++ def request(self, replica, sql, settings=None, user='app'):+ settings = {**dict(default_format='JSONEachRow', wait_end_of_query=1, async_insert=0, mutations_sync=2, alter_sync=2, fsync_metadata=1,+ max_result_rows=0, max_result_bytes=0, result_overflow_mode='throw', read_overflow_mode='throw',+ distributed_ddl_task_timeout=5, replication_wait_for_inactive_replica_timeout=5, insert_quorum_timeout=5000), **(settings or {})}+ connection = http.client.HTTPConnection('127.0.0.1', self.replicas[replica].port, timeout=15)+ auth = 'Basic ' + base64.b64encode((user + ':' + PASSWORD).encode()).decode()+ started = time.monotonic()+ try:+ connection.request('POST', '/?' + urllib.parse.urlencode(settings), sql.encode(), {'Authorization': auth})+ response = connection.getresponse()+ body = response.read().decode()+ headers = dict(response.getheaders())+ return dict(endpoint=self.replicas[replica].endpoint, replica='r' + str(replica + 1), sql=sql, settings=settings, status=response.status, headers=headers, body=body, elapsedSeconds=round(time.monotonic() - started, 3))+ finally:+ connection.close()++ def query(self, sql):+ return self.replicas[0].query(sql)++ def stop_replica(self, replica):+ self.replicas[replica].stop(signal.SIGKILL)++ def start_replica(self, replica):+ target = self.replicas[replica]+ target.restart()+ deadline = time.monotonic() + 30+ while time.monotonic() < deadline:+ try:+ databases = target.query('SELECT is_readonly FROM system.database_replicas')+ tables = target.query('SELECT is_readonly FROM system.replicas')+ if databases and all(row['is_readonly'] == 0 for row in databases + tables):+ return+ except OSError:+ pass+ time.sleep(.05)+ raise RuntimeError('Restarted replicated databases remained read-only; fixture failed without restart retry')++ def close(self, success):+ if self.closed:+ return+ self.closed = True+ if self.proxy:+ self.proxy.close()+ for replica in reversed(self.replicas):+ replica.close(success)+ if self.keeper:+ logs = subprocess.run(self.docker + ['logs', self.keeper], capture_output=True, text=True)+ (self.root / 'keeper-container.log').write_text(logs.stdout + logs.stderr)+ subprocess.run(self.docker + ['rm', '-f', self.keeper], capture_output=True, timeout=15)+ if hasattr(self, 'keeper_process'):+ self.keeper_process.wait(timeout=10)+ self.keeper_log.close()+ subprocess.run(self.docker + ['network', 'rm', self.network], capture_output=True, timeout=15)+ if success:+ shutil.rmtree(self.root)+ else:+ print('Replicated fixture evidence: ' + str(self.root), file=__import__('sys').stderr, flush=True)