diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/clickhouse-migrate-test-support.cabal b/clickhouse-migrate-test-support.cabal --- a/clickhouse-migrate-test-support.cabal +++ b/clickhouse-migrate-test-support.cabal @@ -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, diff --git a/src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs b/src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs --- a/src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs +++ b/src/ClickHouseMigrate/TestSupport/Internal/Fixture.hs @@ -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 diff --git a/src/ClickHouseMigrate/TestSupport/Server.hs b/src/ClickHouseMigrate/TestSupport/Server.hs --- a/src/ClickHouseMigrate/TestSupport/Server.hs +++ b/src/ClickHouseMigrate/TestSupport/Server.hs @@ -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]) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/ReplicatedIntegration.hs b/test/ReplicatedIntegration.hs new file mode 100644 --- /dev/null +++ b/test/ReplicatedIntegration.hs @@ -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 :| [])) diff --git a/test/data/clickhouse-matrix.json b/test/data/clickhouse-matrix.json --- a/test/data/clickhouse-matrix.json +++ b/test/data/clickhouse-matrix.json @@ -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" } } ], diff --git a/test/scripts/native_fixture.py b/test/scripts/native_fixture.py --- a/test/scripts/native_fixture.py +++ b/test/scripts/native_fixture.py @@ -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 @@ defaultdefault1 360000 000''') - 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('127.0.0.1', '::/0')) + 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''' warning{root}/server.log{root}/error.log {root}/data/{root}/tmp/{root}/user_files/ -{root}/format_schemas/{root}/users.xml -127.0.0.1{self.port}{self.tcp_port} +{root}/access/ +{root}/format_schemas/{users_path} +{listener}{http_port}{tcp_port} 64128 82 2 @@ -90,8 +113,28 @@ 22 00 ''') + if deployment: + extra = deployment['config'] + config = self.root / 'config.xml' + config.write_text(config.read_text().replace('', extra + '')) 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) diff --git a/test/scripts/replicated_fixture.py b/test/scripts/replicated_fixture.py new file mode 100644 --- /dev/null +++ b/test/scripts/replicated_fixture.py @@ -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('''warningtrue +0.0.0.091811 +/fixture/log/fixture/snapshots +1000010000 +1localhost9234 +''') + 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 = '' + ''.join('' + name + '9000default' + PASSWORD + '' for name in names) + '' + for i, name in enumerate(names): + extra = cluster + '' + keeper_name + '9181fixture1r' + str(i + 1) + '9009' + name + '/clickhouse/task_queue/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)