diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,14 @@
# Revision history for salmon-ops-recipes
-## 0.2.0.0 -- unreleased
+## 0.3.0.0 -- 2026-10-31
+
+No change to the library's own code. Requires `salmon-ops ^>=0.3.0.0`, whose exports changed.
+
+* Every external dependency now carries a PVP upper bound.
+* Tests for the ClickHouse, PeerDB and `run serve` start-up/`--once`/`--json-log` additions of
+ `salmon-ops` (Layer 0; Layer 2 under podman, PeerDB opt-in with `SALMON_TEST_PEERDB=1`).
+
+## 0.2.0.0 -- 2026-10-23
Breaking: `SreBox.PostgresPair`'s records and some exported functions changed. Requires
`salmon-ops ^>=0.2.0.0`.
diff --git a/salmon-ops-recipes.cabal b/salmon-ops-recipes.cabal
--- a/salmon-ops-recipes.cabal
+++ b/salmon-ops-recipes.cabal
@@ -1,6 +1,6 @@
cabal-version: 2.4
name: salmon-ops-recipes
-version: 0.2.0.0
+version: 0.3.0.0
synopsis: Opinionated recipes built from Salmon builtins.
description: Higher-level recipes (postgres migrations, postgres primary/standby pairs, micro DNS, ...) composed from salmon-ops builtins.
homepage: https://lucasdicioccio.github.io/salmon/
@@ -45,37 +45,37 @@
, SreBox.Gcp.PreviewEnvironment
, SreBox.Gcp.PlakarGcsBackup
build-depends: base >=4.16.3.0 && <4.22
- , aeson
- , base16-bytestring
- , base64-bytestring
- , bytestring
- , case-insensitive
- , comonad
- , containers
- , contravariant
- , cryptohash-sha256
- , crypton-connection
- , crypton-x509
- , crypton-x509-store
- , crypton-x509-validation
- , directory
- , filepath
- , free
- , hashable
- , http-client
- , http-client-tls
- , jose
- , jose-jwt
- , mtl
- , optparse-applicative
- , optparse-generic
- , process
- , process-extras
+ , aeson ^>=2.3.2.0
+ , base16-bytestring ^>=1.0.2.0
+ , base64-bytestring ^>=1.2.1.0
+ , bytestring ^>=0.12.2.0
+ , case-insensitive ^>=1.2.1.0
+ , comonad ^>=5.0.10
+ , containers ^>=0.7
+ , contravariant ^>=1.5.6
+ , cryptohash-sha256 ^>=0.11.102.1
+ , crypton-connection ^>=0.4.6
+ , crypton-x509 ^>=1.10.0
+ , crypton-x509-store ^>=1.10.0
+ , crypton-x509-validation ^>=1.10.0
+ , directory ^>=1.3.8.5
+ , filepath ^>=1.5.4.0
+ , free ^>=5.2
+ , hashable ^>=1.5.1.0
+ , http-client ^>=0.7.19
+ , http-client-tls ^>=0.4.0
+ , jose ^>=0.13
+ , jose-jwt ^>=0.11.0
+ , mtl ^>=2.3.1
+ , optparse-applicative ^>=0.19.0.0
+ , optparse-generic ^>=1.5.3
+ , process ^>=1.6.26.1
+ , process-extras ^>=0.7.4
, salmon-core ^>=0.1.0.0
- , salmon-ops ^>=0.2.0.0
- , text
- , time
- , tls
+ , salmon-ops ^>=0.3.0.0
+ , text ^>=2.1.3
+ , time ^>=1.12.2
+ , tls ^>=2.4.12
hs-source-dirs: src
default-language: Haskell2010
default-extensions: KindSignatures
@@ -152,6 +152,8 @@
, Test.PostgresInitSpec
, Test.PostgresReplicationSpec
, Test.PostgresClusterSpec
+ , Test.ClickHouseSpec
+ , Test.PeerDBSpec
, Test.PgBouncerSpec
, Test.HaproxySpec
, Test.EtcdSpec
@@ -180,6 +182,7 @@
, Test.RewriteSpec
, Test.ServeEventsSpec
, Test.ServeHttpSpec
+ , Test.ServeJsonLogSpec
, Test.ServeModelSpec
, Test.ServeSocketSpec
, Test.ServeSpec
@@ -197,41 +200,41 @@
, Test.UpkeepRetentionSpec
, Test.WindowSpec
, Test.WrapperOrderingSpec
- build-depends: QuickCheck
+ build-depends: QuickCheck ^>=2.19.0.0
, base >=4.16.3.0 && <4.22
- , aeson
- , bytestring
- , comonad
- , containers
- , contravariant
- , crypton-connection
- , crypton-x509-store
- , directory
- , filepath
- , free
- , ghc-heap
- , http-client
- , http-client-tls
- , http-types
- , mtl
- , network
- , optparse-applicative
- , process
- , process-extras
- , stm
+ , aeson ^>=2.3.2.0
+ , bytestring ^>=0.12.2.0
+ , comonad ^>=5.0.10
+ , containers ^>=0.7
+ , contravariant ^>=1.5.6
+ , crypton-connection ^>=0.4.6
+ , crypton-x509-store ^>=1.10.0
+ , directory ^>=1.3.8.5
+ , filepath ^>=1.5.4.0
+ , free ^>=5.2
+ , ghc-heap ^>=9.10.3
+ , http-client ^>=0.7.19
+ , http-client-tls ^>=0.4.0
+ , http-types ^>=0.12.6
+ , mtl ^>=2.3.1
+ , network ^>=3.2.9.0
+ , optparse-applicative ^>=0.19.0.0
+ , process ^>=1.6.26.1
+ , process-extras ^>=0.7.4
+ , stm ^>=2.5.3.1
, salmon-core ^>=0.1.0.0
- , salmon-ops ^>=0.2.0.0
- , salmon-ops-recipes ^>=0.2.0.0
- , tasty
- , tasty-hunit
- , tasty-quickcheck
- , temporary
- , text
- , time
- , tls
- , unix
- , wai
- , warp
+ , salmon-ops ^>=0.3.0.0
+ , salmon-ops-recipes ^>=0.3.0.0
+ , tasty ^>=1.5.5
+ , tasty-hunit ^>=0.10.2
+ , tasty-quickcheck ^>=0.11.1
+ , temporary ^>=1.3
+ , text ^>=2.1.3
+ , time ^>=1.12.2
+ , tls ^>=2.4.12
+ , unix ^>=2.8.7.0
+ , wai ^>=3.2.5
+ , warp ^>=3.4.16
-- Salmon.Actions.Serve reads its input handle on a forked thread, so a
-- test driving the loop wants a runtime where a blocking read does not
-- stall every other Haskell thread. -rtsopts so that a large-graph run
@@ -257,20 +260,20 @@
, Test.DagReference
, Test.PathListReference
build-depends: base >=4.16.3.0 && <4.22
- , aeson
- , bytestring
- , comonad
- , containers
- , free
- , mtl
- , optparse-applicative
- , process
- , stm
- , unix
+ , aeson ^>=2.3.2.0
+ , bytestring ^>=0.12.2.0
+ , comonad ^>=5.0.10
+ , containers ^>=0.7
+ , free ^>=5.2
+ , mtl ^>=2.3.1
+ , optparse-applicative ^>=0.19.0.0
+ , process ^>=1.6.26.1
+ , stm ^>=2.5.3.1
+ , unix ^>=2.8.7.0
, salmon-core ^>=0.1.0.0
- , salmon-ops ^>=0.2.0.0
- , text
- , time
+ , salmon-ops ^>=0.3.0.0
+ , text ^>=2.1.3
+ , time ^>=1.12.2
-- one capability, as the binaries in salmon-apps are built; -T for the
-- allocation counter; -rtsopts so a run can be given a heap bound.
ghc-options: -threaded -rtsopts "-with-rtsopts=-T"
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -55,6 +55,8 @@
import qualified Test.PostgresInitSpec as PostgresInitSpec
import qualified Test.PostgresReplicationSpec as PostgresReplicationSpec
import qualified Test.PostgresClusterSpec as PostgresClusterSpec
+import qualified Test.ClickHouseSpec as ClickHouseSpec
+import qualified Test.PeerDBSpec as PeerDBSpec
import qualified Test.PgBouncerSpec as PgBouncerSpec
import qualified Test.HaproxySpec as HaproxySpec
import qualified Test.EtcdSpec as EtcdSpec
@@ -81,6 +83,7 @@
import qualified Test.RewriteSpec as RewriteSpec
import qualified Test.ServeEventsSpec as ServeEventsSpec
import qualified Test.ServeHttpSpec as ServeHttpSpec
+import qualified Test.ServeJsonLogSpec as ServeJsonLogSpec
import qualified Test.ServeModelSpec as ServeModelSpec
import qualified Test.ServeSocketSpec as ServeSocketSpec
import qualified Test.ServeSpec as ServeSpec
@@ -134,6 +137,7 @@
, PgBackRestSpec.sandboxTests
, PgTextSearchSpec.sandboxTests
, PgVectorDemoSpec.sandboxTests
+ , PeerDBSpec.sandboxTests
, -- the rest of the VM specs: they share one bridge and a
-- handful of fixed addresses, so two at once is two guests
-- claiming one address.
@@ -199,6 +203,8 @@
, PodmanSpec.tests
, PostgresBackupSpec.tests
, PostgresClusterSpec.tests
+ , ClickHouseSpec.tests
+ , PeerDBSpec.tests
, PgBouncerSpec.tests
, HaproxySpec.tests
, EtcdSpec.tests
@@ -221,6 +227,7 @@
, RewriteSpec.tests
, ServeEventsSpec.tests
, ServeHttpSpec.tests
+ , ServeJsonLogSpec.tests
, ServeModelSpec.tests
, ServeSocketSpec.tests
, ServeSpec.tests
diff --git a/test/Test/ClickHouseSpec.hs b/test/Test/ClickHouseSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Test/ClickHouseSpec.hs
@@ -0,0 +1,205 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+{- | "Salmon.Builtin.Nodes.ClickHouse".
+
+Layer 0: the drop-ins, the quadlet, the SQL each node sends, and the
+readings of what the server answers. Layer 2 ('liveServer'): the official
+image run under podman with the rendered drop-ins mounted, and the database,
+user and grant nodes run against it through @podman exec@, twice, then with
+a rotated password. The quadlet itself ('ClickHouse.serverOp') is Layer 0
+only here: it is a system-scope unit.
+-}
+module Test.ClickHouseSpec (tests) where
+
+import Control.Exception (bracket)
+import Control.Monad (forM_, void)
+import qualified Data.ByteString as ByteString
+import Data.List (isInfixOf)
+import qualified Data.Text as Text
+import System.Exit (ExitCode (..))
+import System.FilePath (())
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)
+
+import Salmon.Actions.UpDown (CheckResult (..))
+import qualified Salmon.Actions.UpDown as UpDown
+import Salmon.Builtin.Extension
+import Salmon.Op.Actions (Act (..))
+import Salmon.Op.OpGraph (inject)
+import qualified Salmon.Builtin.Nodes.ClickHouse as ClickHouse
+import qualified Salmon.Builtin.Nodes.Filesystem as FS
+import qualified Salmon.Builtin.Nodes.Podman as Podman
+import qualified Salmon.Builtin.Nodes.Podman.Quadlet as Quadlet
+import qualified Salmon.Builtin.Nodes.Readiness as Readiness
+import Salmon.Reporter (silent)
+import Salmon.Op.Track (Track (..))
+import Test.Harness
+
+tests :: TestTree
+tests =
+ testGroup
+ "Salmon.Builtin.Nodes.ClickHouse"
+ [ testGroup "Layer 0" layer0
+ , testCase "Layer 2: database, user and grant nodes against the official image under podman" liveServer
+ ]
+
+layer0 :: [TestTree]
+layer0 =
+ [ testCase "the default configuration renders the root element and nothing else" $
+ assertEqual "" "\n\n" (ClickHouse.renderServerConfig ClickHouse.defaultServerConfig)
+ , testCase "listen hosts, ports and settings are rendered, values escaped" $ do
+ let cfg =
+ ClickHouse.defaultServerConfig
+ { ClickHouse.serverListenHosts = ["0.0.0.0"]
+ , ClickHouse.serverTcpPort = Just 9440
+ , ClickHouse.serverHttpPort = Just 8124
+ , ClickHouse.serverSettings = [("display_name", "a0.0.0.0"
+ , "9440"
+ , "8124"
+ , "a<b & 'c'"
+ ]
+ $ \line -> assertBool (line <> " in\n" <> out) (line `isInfixOf` out)
+ , testCase "a setting that is not a plain element name is refused" $ do
+ let bad = ClickHouse.defaultServerConfig{ClickHouse.serverSettings = [("a>1" `isInfixOf` out)
+ assertBool out ("" `isInfixOf` out)
+ assertBool out ("127.0.0.1" `isInfixOf` out && "::1" `isInfixOf` out)
+ assertBool out (not ("::/0" `isInfixOf` out))
+ , testCase "identifiers and literals are quoted with their quotes escaped" $ do
+ assertEqual "" "`we\\`ird`" (ClickHouse.quoteIdent "we`ird")
+ assertEqual "" "'it\\'s \\\\ x'" (ClickHouse.quoteLiteral "it's \\ x")
+ , testCase "a database is created if missing" $
+ assertEqual "" "CREATE DATABASE IF NOT EXISTS `analytics`;\n" (ClickHouse.createDatabaseSql "analytics")
+ , testCase "ClickHouse's sha256_hash is the SHA-256 of password then salt" $
+ -- printf 's3cretabcd' | sha256sum; and ClickHouse 25.8 accepted
+ -- the password once this was set (see liveServer)
+ assertEqual "" "35496151d4e1741c05ddca164851a963765345081a76900479fb388696bc073d" (ClickHouse.sha256Hash (ClickHouse.Password "s3cret") "abcd")
+ , testCase "the user statements create then alter, with the hash and never the password" $ do
+ let sql = Text.unpack (ClickHouse.userSql "peer" (ClickHouse.Password "s3cret") (ByteString.pack [0xab, 0xcd]))
+ assertBool sql ("CREATE USER IF NOT EXISTS `peer` IDENTIFIED WITH sha256_hash BY '35496151d4e1741c05ddca164851a963765345081a76900479fb388696bc073d' SALT 'abcd';" `isInfixOf` sql)
+ assertBool sql ("ALTER USER `peer` IDENTIFIED WITH sha256_hash" `isInfixOf` sql)
+ assertBool sql (not ("s3cret" `isInfixOf` sql))
+ , testCase "redaction takes the hash and the salt out of client output" $
+ assertEqual "" "failed at '' SALT ''" (ClickHouse.redactAll ["deadbeef", "abcd"] "failed at 'deadbeef' SALT 'abcd'")
+ , testCase "a password-free help, notes and argv for the user node" $ do
+ let c = ClickHouse.Client (ClickHouse.InContainer (Podman.ContainerName "ch")) 9000
+ argv = ClickHouse.loginArgv c "peer"
+ assertEqual "" ["podman", "exec", "-i", "-e", "CLICKHOUSE_PASSWORD", "ch", "clickhouse-client", "--port", "9000", "--user", "peer", "--query", "SELECT currentUser()"] argv
+ let o = ClickHouse.userPassFile silent ignoreTrack c (FS.PreExisting "/run/secrets/ch-peer") "peer"
+ case opAct o of
+ Just a -> assertBool "the help names the file" ("/run/secrets/ch-peer" `Text.isInfixOf` a.extension.help)
+ Nothing -> assertFailure "an actionless node"
+ , testCase "the counts read" $ do
+ assertEqual "" Success (ClickHouse.interpretCount "x" "1\n")
+ assertEqual "" (Failure "x") (ClickHouse.interpretCount "x" "0\n")
+ assertEqual "" Unknown (ClickHouse.interpretCount "x" "")
+ , testCase "a refused login is a Failure, anything else that fails is Unknown" $ do
+ assertEqual "" Success (ClickHouse.interpretLogin "peer" ExitSuccess "")
+ assertBool "refused" $ case ClickHouse.interpretLogin "peer" (ExitFailure 4) "Code: 516. DB::Exception: peer: Authentication failed (AUTHENTICATION_FAILED)" of
+ Failure _ -> True
+ _ -> False
+ assertEqual "" Unknown (ClickHouse.interpretLogin "peer" (ExitFailure 210) "Connection refused (NETWORK_ERROR)")
+ , testCase "privileges are compared without case, and one missing is named" $ do
+ let g = ClickHouse.Grant ["select", "create table", "dictGet"] "analytics" "peer"
+ assertEqual "" "GRANT SELECT, CREATE TABLE, DICTGET ON `analytics`.* TO `peer`;\n" (ClickHouse.grantSql g)
+ assertEqual "" Success (ClickHouse.interpretGrants g "SELECT\nINSERT\nCREATE TABLE\ndictGet\n")
+ assertEqual "" (Failure "peer lacks CREATE TABLE on analytics.*") (ClickHouse.interpretGrants g "SELECT\ndictGet\n")
+ , testCase "the quadlet mounts data and drop-ins, watches the drop-ins, and waits for SELECT 1" $ do
+ let s = ClickHouse.server (Podman.ContainerName "ch") "docker.io/clickhouse/clickhouse-server:25.8" "/etc/salmon/ch" "/var/lib/salmon/ch"
+ c = ClickHouse.serverContainer s
+ out = Text.unpack (Quadlet.renderContainer c)
+ assertEqual "no problems" [] (Quadlet.containerProblems c)
+ forM_
+ [ "Image=docker.io/clickhouse/clickhouse-server:25.8"
+ , "Volume=/var/lib/salmon/ch:/var/lib/clickhouse:rw"
+ , "Volume=/etc/salmon/ch/salmon-server.xml:/etc/clickhouse-server/config.d/salmon-server.xml:ro"
+ , "Volume=/etc/salmon/ch/salmon-admin.xml:/etc/clickhouse-server/users.d/salmon-admin.xml:ro"
+ ]
+ $ \line -> assertBool (line <> " in\n" <> out) (line `isInfixOf` out)
+ assertEqual "" ["/etc/salmon/ch/salmon-server.xml", "/etc/salmon/ch/salmon-admin.xml"] (Quadlet.watchedFiles c)
+ assertEqual "" (Just (Quadlet.readyWhen (ClickHouse.readyProbe (ClickHouse.serverClient s)) 60)) (Quadlet.containerReady c)
+ ]
+
+-------------------------------------------------------------------------------
+
+{- | The official image, the rendered drop-ins mounted, the nodes run twice
+(the second pass skips all three on their checks), the password rotated
+(the user check fails, @up@ puts it back), and @default@ allowed from the
+local machine only (the admin drop-in did apply).
+-}
+liveServer :: IO ()
+liveServer = requireExecutable "podman" $
+ withTempDir $ \dir -> do
+ let serverXml = dir "salmon-server.xml"
+ adminXml = dir "salmon-admin.xml"
+ passfile = dir "peer.pass"
+ writeFile serverXml (Text.unpack (ClickHouse.renderServerConfig ClickHouse.defaultServerConfig{ClickHouse.serverSettings = [("max_connections", "512")]}))
+ writeFile adminXml (Text.unpack ClickHouse.renderAdminAccess)
+ writeFile passfile "first-password"
+ cname <- freshContainerName
+ (reporter, readBack) <- capture
+ let img = Podman.Image "clickhouse/clickhouse-server:25.8"
+ opts =
+ Podman.noRunOptions
+ { Podman.runVolumes =
+ [ Podman.VolumeMount serverXml "/etc/clickhouse-server/config.d/salmon-server.xml" Podman.ReadOnly
+ , Podman.VolumeMount adminXml "/etc/clickhouse-server/users.d/salmon-admin.xml" Podman.ReadOnly
+ ]
+ }
+ runOp = Podman.runContainer reporter podmanTrack Podman.dockerRegistry img cname opts
+ pullOp = Podman.pullImage reporter podmanTrack Podman.dockerRegistry img
+ container = Text.unpack (Podman.getContainerName cname)
+ pulled <- runUp pullOp
+ assertBool "pull the image" pulled
+ bracket (runUp runOp) (\_ -> void (runDown runOp)) $ \started -> do
+ failures <- podmanFailures <$> readBack
+ assertBool ("start the container" <> concat failures) started
+ let client = ClickHouse.Client (ClickHouse.InContainer cname) ClickHouse.defaultTcpPort
+ waitOp = Readiness.waitUntil (Readiness.untilProbe "clickhouse-test-ready" (ClickHouse.readyProbe client) 60)
+ waitT = Track (const waitOp)
+ ready <- runUp waitOp
+ assertBool "the server answers within 60s" ready
+ (chr, _) <- capture
+ let grant = ClickHouse.Grant ["SELECT", "INSERT", "CREATE TABLE", "ALTER"] "analytics" "peer"
+ dbOp = ClickHouse.database chr waitT client "analytics"
+ userOp = ClickHouse.userPassFile chr waitT client (FS.PreExisting passfile) "peer"
+ -- a grant needs its user and its database: edges, as in a recipe
+ grantOp = (ClickHouse.grant chr waitT client grant `inject` userOp) `inject` dbOp
+ graph = op "clickhouse-test" (deps [grantOp]) id
+ first <- runUpCapturing graph
+ assertBool ("first pass succeeds: " <> show (failed first)) (null (failed first))
+ assertEqual "database" Success =<< ClickHouse.checkDatabase client "analytics"
+ assertEqual "user" Success =<< ClickHouse.checkUser client "peer" (ClickHouse.readPassword passfile)
+ assertEqual "grant" Success =<< ClickHouse.checkGrant client grant
+ second <- runUpCapturing graph
+ assertEqual "the second pass applies nothing" [] [showAct a | UpDown.Done a <- second, isClickHouse a]
+ writeFile passfile "rotated-password"
+ rotated <- ClickHouse.checkUser client "peer" (ClickHouse.readPassword passfile)
+ assertBool ("a rotated file is a Failure: " <> show rotated) (isFailure rotated)
+ third <- runUpCapturing graph
+ assertBool ("third pass succeeds: " <> show (failed third)) (null (failed third))
+ assertEqual "the rotated password logs in" Success =<< ClickHouse.checkUser client "peer" (ClickHouse.readPassword passfile)
+ -- default is loopback-only: ClickHouse folds ::1 and 127.0.0.1
+ -- into "localhost" (it counts every address of the machine as
+ -- local, so a client inside the container on its own address
+ -- is let in; a remote one is not, which is not tried here)
+ (_, hostsOut, _) <- podmanExecCapture container ["clickhouse-client", "--query", "SELECT host_ip, host_names FROM system.users WHERE name = 'default' FORMAT TSV"]
+ assertEqual "default's allowed hosts" "[]\t['localhost']" (concat (lines hostsOut))
+ -- the hash never reaches the query log as written
+ (_, logOut, _) <- podmanExecCapture container ["clickhouse-client", "--query", "SYSTEM FLUSH LOGS; SELECT query FROM system.query_log WHERE query ILIKE '%IDENTIFIED%'", "--multiquery"]
+ assertBool logOut (not ("sha256_hash BY '" `isInfixOf` logOut))
+ where
+ showAct :: Act' -> String
+ showAct a = Text.unpack a.extension.help
+ isClickHouse :: Act' -> Bool
+ isClickHouse a = "ClickHouse" `Text.isInfixOf` a.extension.help || "grant " `Text.isPrefixOf` a.extension.help
+ failed reports = [showAct a <> ": " <> show e | UpDown.Failed a e <- reports]
+ isFailure (Failure _) = True
+ isFailure _ = False
diff --git a/test/Test/Harness.hs b/test/Test/Harness.hs
--- a/test/Test/Harness.hs
+++ b/test/Test/Harness.hs
@@ -43,6 +43,7 @@
-- * redirecting a recipe's system binaries into a container via PATH shims
withShimmedPath,
+ withPrependedPath,
-- * qemu-backed sandboxes (Layer 3)
testBridge,
diff --git a/test/Test/PeerDBSpec.hs b/test/Test/PeerDBSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Test/PeerDBSpec.hs
@@ -0,0 +1,541 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+{- | "Salmon.Builtin.Nodes.PeerDB", "Salmon.Builtin.Nodes.Postgres.Logical"
+and 'ClickHouse.sourceReadGrant': CDC from PostgreSQL into ClickHouse.
+
+Layer 0: the SQL the Postgres nodes send, the bodies sent to PeerDB's flow
+API (and that the password is in the peer's body only), and the readings of
+what PostgreSQL and PeerDB answer, on answers copied from PeerDB
+@stable-v0.37.12@.
+
+Layer 2 ('sandboxTests', opt-in with @SALMON_TEST_PEERDB=1@): PeerDB's
+services as upstream's compose file runs them at @stable-v0.37.12@ (catalog,
+Temporal, the object store, @flow-api@ and both workers; not the UI nor the
+pgwire server), a PostgreSQL 17 source and a ClickHouse 25.8 target, all on
+one podman network, started by plain @podman run@ here. The nodes run
+against them: the Postgres ones as the @postgres@ OS user through a @sudo@
+shim into the source container.
+-}
+module Test.PeerDBSpec (tests, sandboxTests) where
+
+import Control.Concurrent (threadDelay)
+import Control.Exception (SomeException, bracket_, try)
+import Control.Monad (forM_, unless, void, when)
+import qualified Data.Aeson as Aeson
+import qualified Data.ByteString.Lazy.Char8 as LChar8
+import Data.List (isInfixOf)
+import Data.Text (Text)
+import qualified Data.Text as Text
+import System.Environment (lookupEnv)
+import System.Exit (ExitCode (..))
+import System.FilePath (())
+import System.IO (hPutStrLn, stderr)
+import System.Process (readProcessWithExitCode)
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)
+
+import Salmon.Actions.UpDown (CheckResult (..))
+import qualified Salmon.Actions.UpDown as UpDown
+import Salmon.Builtin.Extension
+import qualified Salmon.Builtin.Nodes.ClickHouse as ClickHouse
+import qualified Salmon.Builtin.Nodes.Filesystem as FS
+import qualified Salmon.Builtin.Nodes.PeerDB as PeerDB
+import qualified Salmon.Builtin.Nodes.Podman as Podman
+import qualified Salmon.Builtin.Nodes.Postgres as Postgres
+import qualified Salmon.Builtin.Nodes.Postgres.Logical as Logical
+import Salmon.Op.Actions (Act (..), Actions (..))
+import Salmon.Op.OpGraph (OpGraph (..), inject)
+
+import Salmon.Reporter (silent)
+import Test.Harness
+
+tests :: TestTree
+tests =
+ testGroup
+ "Salmon.Builtin.Nodes.PeerDB and Postgres.Logical (Layer 0)"
+ [ testGroup "Postgres.Logical" logicalTests
+ , testGroup "PeerDB" peerdbTests
+ , testCase "ClickHouse: READ ON S3, and its check" $ do
+ assertEqual "" "GRANT READ ON S3 TO `peer`;\n" (ClickHouse.sourceReadGrantSql "s3" "peer")
+ assertEqual "a source name is upper-cased" "GRANT READ ON URL TO `peer`;\n" (ClickHouse.sourceReadGrantSql "url" "peer")
+ assertEqual "an odd source name is not markup" "GRANT READ ON S3DROP TO `peer`;\n" (ClickHouse.sourceReadGrantSql "s3; DROP" "peer")
+ assertBool "" ("access_object = 'S3'" `Text.isInfixOf` ClickHouse.sourceReadGrantedSql "s3" "peer")
+ ]
+
+pub :: Logical.Publication
+pub = Logical.Publication "cdc_pub" "app" [Logical.publicTable "orders", Logical.QualifiedTable "Sales" "line \"items\""]
+
+logicalTests :: [TestTree]
+logicalTests =
+ [ testCase "wal_level: set, reloaded; the check reads the files, then the running value" $ do
+ assertEqual "" "ALTER SYSTEM SET wal_level = 'logical';\nSELECT pg_reload_conf();\n" Logical.walLevelSql
+ assertEqual "" Success (Logical.interpretWalLevel "logical\n")
+ assertEqual "" (Failure "wal_level is replica, logical decoding needs logical") (Logical.interpretWalLevel "replica")
+ assertEqual "" Unknown (Logical.interpretWalLevel "")
+ , testCase "the wal_level node is on alterSystemSet's effect site" $ do
+ let mine = nodeOf (Logical.logicalWalLevel silent ignoreTrack 5432)
+ theirs = nodeOf (Postgres.alterSystemSet silent ignoreTrack 5432 "wal_level" "replica")
+ assertEqual "" theirs.ref mine.ref
+ , testCase "REPLICATION: the statement and the verdicts" $ do
+ assertEqual "" "ALTER ROLE \"cdc\" WITH REPLICATION;\n" (Logical.replicationRoleSql "cdc")
+ assertEqual "" Success (Logical.interpretReplicationRole "cdc" "t\n")
+ assertEqual "" (Failure "role cdc may not replicate") (Logical.interpretReplicationRole "cdc" "f")
+ assertEqual "" (Failure "role cdc does not exist") (Logical.interpretReplicationRole "cdc" "")
+ , testCase "a publication: refusal of an unmarked one, create via gexec, set table, marker, one transaction" $ do
+ let sql = Logical.publicationSql pub
+ ls = Text.lines sql
+ assertEqual "first and last" (["BEGIN;"], ["COMMIT;"]) (take 1 ls, drop (length ls - 1) ls)
+ forM_
+ [ "NOT LIKE 'salmon-publication:%'"
+ , "RAISE EXCEPTION"
+ , "SELECT 'CREATE PUBLICATION \"cdc_pub\" FOR TABLE \"public\".\"orders\", \"Sales\".\"line \"\"items\"\"\"' WHERE NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'cdc_pub') \\gexec"
+ , "ALTER PUBLICATION \"cdc_pub\" SET TABLE \"public\".\"orders\", \"Sales\".\"line \"\"items\"\"\";"
+ , "COMMENT ON PUBLICATION \"cdc_pub\" IS 'salmon-publication: app';"
+ ]
+ $ \frag -> assertBool (Text.unpack frag <> " in\n" <> Text.unpack sql) (frag `Text.isInfixOf` sql)
+ , testCase "a publication's verdicts" $ do
+ let marker = "present:salmon-publication: app"
+ tables = "Sales.line \"items\"\npublic.orders\n"
+ assertEqual "" Success (Logical.interpretPublication pub marker tables)
+ assertBool "absent" (isFailure (Logical.interpretPublication pub "" ""))
+ assertEqual
+ "unmarked"
+ (Failure "publication cdc_pub exists but salmon did not make it")
+ (Logical.interpretPublication pub "present:" tables)
+ assertEqual
+ "other tables"
+ (Failure "publication cdc_pub publishes public.orders, declared Sales.line \"items\", public.orders")
+ (Logical.interpretPublication pub marker "public.orders")
+ , testCase "a logical slot: refused when the name is taken by another kind, created when missing" $ do
+ let sql = Logical.logicalSlotSql "app" "cdc_slot"
+ assertBool (Text.unpack sql) ("pg_create_logical_replication_slot('cdc_slot', 'pgoutput') WHERE NOT EXISTS" `Text.isInfixOf` sql)
+ assertBool (Text.unpack sql) ("slot_type = 'logical' AND plugin = 'pgoutput' AND database = 'app'" `Text.isInfixOf` sql)
+ assertEqual "" Success (Logical.interpretSlot "app" "cdc_slot" "logical|pgoutput|app\n")
+ assertEqual "" (Failure "replication slot cdc_slot does not exist") (Logical.interpretSlot "app" "cdc_slot" "")
+ assertBool "physical" (isFailure (Logical.interpretSlot "app" "cdc_slot" "physical||"))
+ , testCase "the slot node is on the physical slot node's effect site" $
+ assertEqual "" (nodeOf (Postgres.replicationSlot silent ignoreTrack 5432 "s")).ref (nodeOf (Logical.logicalSlot silent ignoreTrack 5432 "app" "s")).ref
+ , testCase "nothing here has a down" $
+ forM_
+ [ Logical.logicalWalLevel silent ignoreTrack 5432
+ , Logical.replicationRole silent ignoreTrack 5432 "cdc"
+ , Logical.publication silent ignoreTrack 5432 pub
+ , Logical.logicalSlot silent ignoreTrack 5432 "app" "s"
+ ]
+ $ \o -> do
+ ok <- runDown o
+ assertBool "down does nothing and succeeds" ok
+ ]
+
+srcPeer :: PeerDB.Peer
+srcPeer = PeerDB.Peer "pg_source" (PeerDB.PostgresSource (PeerDB.PostgresPeer "db.internal" 5432 "cdc" "app" False))
+
+dstPeer :: PeerDB.Peer
+dstPeer = PeerDB.Peer "ch_target" (PeerDB.ClickHouseTarget (PeerDB.ClickHousePeer "ch.internal" 9000 "peer" "analytics" True))
+
+theMirror :: PeerDB.Mirror
+theMirror =
+ (PeerDB.mirror "orders_to_ch" "pg_source" "ch_target" [PeerDB.TableMapping "public.orders" "orders"])
+ { PeerDB.mirrorPublication = Just "cdc_pub"
+ , PeerDB.mirrorIdleTimeoutSeconds = Just 5
+ }
+
+-- | What @GET /v1/peers/info/pg_source@ answered at stable-v0.37.12, with this test's names.
+peerInfoAnswer :: Text -> LChar8.ByteString
+peerInfoAnswer host =
+ "{\"peer\":{\"name\":\"pg_source\",\"type\":\"POSTGRES\",\"postgresConfig\":{\"host\":\""
+ <> LChar8.pack (Text.unpack host)
+ <> "\",\"port\":5432,\"user\":\"cdc\",\"password\":\"********\",\"database\":\"app\",\"tlsHost\":\"\",\"requireTls\":false,\"authType\":\"POSTGRES_PASSWORD\",\"skipCertVerification\":false}},\"version\":\"PostgreSQL 17.11\"}"
+
+-- | What @POST /v1/mirrors/status@ answered, trimmed, with this test's names.
+mirrorStatusAnswer :: LChar8.ByteString -> LChar8.ByteString -> LChar8.ByteString
+mirrorStatusAnswer tables state =
+ "{\"flowJobName\":\"orders_to_ch\",\"cdcStatus\":{\"config\":{\"flowJobName\":\"orders_to_ch\",\"tableMappings\":["
+ <> tables
+ <> "],\"publicationName\":\"cdc_pub\",\"sourceName\":\"pg_source\",\"destinationName\":\"ch_target\"},\"sourceType\":\"POSTGRES\",\"destinationType\":\"CLICKHOUSE\",\"rowsSynced\":\"0\"},\"currentFlowState\":\""
+ <> state
+ <> "\",\"createdAt\":\"2026-10-31T07:13:44.731934Z\"}"
+
+ordersMapping :: LChar8.ByteString
+ordersMapping = "{\"sourceTableIdentifier\":\"public.orders\",\"destinationTableIdentifier\":\"orders\",\"engine\":\"CH_ENGINE_REPLACING_MERGE_TREE\"}"
+
+peerdbTests :: [TestTree]
+peerdbTests =
+ [ testCase "a peer's body: the type, the config under its field, allow_update; the password only there" $ do
+ let body = Aeson.encode (PeerDB.createPeerBody dstPeer (PeerDB.Password "s3cr3t"))
+ s = LChar8.unpack body
+ forM_ ["\"type\":\"CLICKHOUSE\"", "\"clickhouse_config\":{", "\"disable_tls\":true", "\"allow_update\":true", "\"password\":\"s3cr3t\""] $ \frag ->
+ assertBool (frag <> " in " <> s) (frag `isInfixOf` s)
+ let o = nodeOf (PeerDB.peer silent ignoreTrack (PeerDB.Api "http://127.0.0.1:8113") (FS.PreExisting "/run/secrets/ch-peer") dstPeer)
+ assertBool "not in help" (not ("s3cr3t" `Text.isInfixOf` o.help))
+ assertBool "the file, in help" ("/run/secrets/ch-peer" `Text.isInfixOf` o.help)
+ , testCase "a peer read back: same is Success, a moved host is a Failure naming it" $ do
+ assertEqual "" (Just Success) (PeerDB.interpretPeerInfo srcPeer <$> Aeson.decode (peerInfoAnswer "db.internal"))
+ assertEqual
+ ""
+ (Just (Failure "PeerDB peer pg_source differs: host is elsewhere, declared db.internal"))
+ (PeerDB.interpretPeerInfo srcPeer <$> Aeson.decode (peerInfoAnswer "elsewhere"))
+ assertBool "a ClickHouse declaration read on a Postgres peer" $
+ maybe False isFailure (PeerDB.interpretPeerInfo dstPeer{PeerDB.peerName = "pg_source"} <$> Aeson.decode (peerInfoAnswer "db.internal"))
+ , testCase "a mirror's body: names, mappings, publication, idle timeout; no slot when none is named" $ do
+ let s = LChar8.unpack (Aeson.encode (PeerDB.createMirrorBody theMirror))
+ forM_
+ [ "\"flow_job_name\":\"orders_to_ch\""
+ , "\"source_table_identifier\":\"public.orders\""
+ , "\"destination_table_identifier\":\"orders\""
+ , "\"publication_name\":\"cdc_pub\""
+ , "\"idle_timeout_seconds\":5"
+ , "\"do_initial_snapshot\":true"
+ ]
+ $ \frag -> assertBool (frag <> " in " <> s) (frag `isInfixOf` s)
+ assertBool s (not ("replication_slot_name" `isInfixOf` s))
+ , testCase "a mirror's status read against the declaration" $ do
+ let verdict tables st = PeerDB.interpretMirrorStatus theMirror <$> Aeson.decode (mirrorStatusAnswer tables st)
+ assertEqual "running" (Just Success) (verdict ordersMapping "STATUS_RUNNING")
+ assertEqual "paused by an operator" (Just Success) (verdict ordersMapping "STATUS_PAUSED")
+ assertEqual "snapshotting" (Just Unknown) (verdict ordersMapping "STATUS_SNAPSHOT")
+ assertEqual "failed" (Just (Failure "PeerDB mirror orders_to_ch is STATUS_FAILED")) (verdict ordersMapping "STATUS_FAILED")
+ assertEqual
+ "other tables"
+ (Just (Failure "PeerDB mirror orders_to_ch differs: tables are none"))
+ (verdict "" "STATUS_RUNNING")
+ , testCase "a declared slot with an initial snapshot is refused; without one, or with PeerDB's own slot, it is not" $ do
+ assertEqual "" [] (PeerDB.mirrorProblems theMirror)
+ assertEqual "" [] (PeerDB.mirrorProblems theMirror{PeerDB.mirrorSlot = Just "s", PeerDB.mirrorInitialSnapshot = False})
+ assertEqual "" 1 (length (PeerDB.mirrorProblems theMirror{PeerDB.mirrorSlot = Just "s"}))
+ assertEqual "" 1 (length (PeerDB.mirrorProblems theMirror{PeerDB.mirrorTables = []}))
+ , testCase "names in a path are escaped" $
+ assertEqual "" "a%2Fb%20c" (PeerDB.pathSegment "a/b c")
+ , testCase "an error's message is kept, the password cut out" $
+ assertEqual
+ ""
+ ": failed for pw "
+ (PeerDB.errorMessage (PeerDB.redact ["hunter2"] "{\"code\":9,\"message\":\"failed for pw hunter2\"}"))
+ ]
+
+nodeOf :: Op -> Extension
+nodeOf o = case o.node of
+ Actions a -> a.extension
+ Actionless -> error "an op without actions"
+
+isFailure :: CheckResult -> Bool
+isFailure (Failure _) = True
+isFailure _ = False
+
+-------------------------------------------------------------------------------
+-- Layer 2
+
+sandboxTests :: TestTree
+sandboxTests =
+ testGroup
+ "Salmon.Builtin.Nodes.PeerDB (Layer 2, PeerDB stable-v0.37.12, PostgreSQL 17 and ClickHouse 25.8 under podman)"
+ [ testCase "prerequisites, peers and a mirror converge, rows stream, a second pass applies nothing" sandbox
+ ]
+
+peerdbTag :: String
+peerdbTag = "stable-v0.37.12"
+
+sandbox :: IO ()
+sandbox = requireExecutable "podman" $ do
+ enabled <- lookupEnv "SALMON_TEST_PEERDB"
+ if enabled /= Just "1"
+ then hPutStrLn stderr "SKIPPED: SALMON_TEST_PEERDB is not 1; this Layer 2 test pulls and runs PeerDB's services (several GB)"
+ else withTempDir $ \dir -> do
+ Podman.ContainerName stem <- freshContainerName
+ apiPort <- freeTcpPort
+ let net = Text.unpack stem
+ name s = net <> "-" <> s
+ bracket_ (pure ()) (teardown net name) $ do
+ stack dir net name apiPort
+ converge dir net name apiPort
+
+{- | Upstream's compose file, by hand: names are prefixed, the object store's
+endpoint is its container, the API is published on a loopback port.
+-}
+stack :: FilePath -> String -> (String -> String) -> Text -> IO ()
+stack dir net name apiPort = do
+ let pgConf = dir "catalog.conf"
+ dynconf = dir "dynamicconfig"
+ writeFile pgConf "listen_addresses = '*'\nwal_level = logical\nmax_wal_senders = 4\nmax_replication_slots = 4\nmax_connections = 2048\n"
+ void (readProcessWithExitCode "mkdir" ["-p", dynconf] "")
+ writeFile (dynconf "development-sql.yaml") "limit.maxIDLength:\n - value: 255\n constraints: {}\nsystem.forceSearchAttributesCacheRefreshOnRead:\n - value: true\n constraints: {}\n"
+ podman ["network", "create", net]
+ let catalog = ["-e", "PEERDB_CATALOG_HOST=" <> name "catalog", "-e", "PEERDB_CATALOG_PORT=5432", "-e", "PEERDB_CATALOG_USER=postgres", "-e", "PEERDB_CATALOG_PASSWORD=postgres", "-e", "PEERDB_CATALOG_DATABASE=postgres"]
+ s3 =
+ [ "-e"
+ , "PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_ACCESS_KEY_ID=_peerdb_minioadmin"
+ , "-e"
+ , "PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_SECRET_ACCESS_KEY=_peerdb_minioadmin"
+ , "-e"
+ , "PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_REGION=us-east-1"
+ , "-e"
+ , "PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_ENDPOINT_URL_S3=http://" <> name "minio" <> ":9000"
+ , "-e"
+ , "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME=peerdbbucket"
+ ]
+ flow = ["-e", "TEMPORAL_HOST_PORT=" <> name "temporal" <> ":7233", "-e", "PEERDB_TEMPORAL_NAMESPACE=default"]
+ detached n args = podman (["run", "-d", "--name", name n, "--network", net] <> args)
+ detached "catalog" ["-e", "POSTGRES_PASSWORD=postgres", "-v", pgConf <> ":/etc/postgresql.conf:ro", "docker.io/library/postgres:17-alpine", "-c", "config_file=/etc/postgresql.conf"]
+ waitUntil "the catalog" 60 (execOk (name "catalog") ["pg_isready", "-U", "postgres"])
+ detached
+ "minio"
+ [ "-e"
+ , "MINIO_ROOT_USER=_peerdb_minioadmin"
+ , "-e"
+ , "MINIO_ROOT_PASSWORD=_peerdb_minioadmin"
+ , "--entrypoint"
+ , "/bin/sh"
+ , "docker.io/pgsty/silo:RELEASE.2026-09-16T00-00-00Z"
+ , "-c"
+ , "silo server /data & sleep 3; mc alias set m http://localhost:9000 _peerdb_minioadmin _peerdb_minioadmin; mc mb m/peerdbbucket; wait"
+ ]
+ detached
+ "temporal"
+ [ "-e"
+ , "DB=postgres12"
+ , "-e"
+ , "DB_PORT=5432"
+ , "-e"
+ , "POSTGRES_USER=postgres"
+ , "-e"
+ , "POSTGRES_PWD=postgres"
+ , "-e"
+ , "POSTGRES_SEEDS=" <> name "catalog"
+ , "-e"
+ , "DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development-sql.yaml"
+ , "-v"
+ , dynconf <> ":/etc/temporal/config/dynamicconfig:ro"
+ , "docker.io/temporalio/auto-setup:1.29"
+ ]
+ podman (["run", "--rm", "--network", net, "--entrypoint", "/home/peerdb/peer-flow"] <> catalog <> ["ghcr.io/peerdb-io/flow-api:" <> peerdbTag, "migrate"])
+ -- what upstream's admin-tools container does at start
+ waitUntil "Temporal and its MirrorName attribute" 180 $
+ ok
+ "podman"
+ [ "run"
+ , "--rm"
+ , "--network"
+ , net
+ , "-e"
+ , "TEMPORAL_ADDRESS=" <> name "temporal" <> ":7233"
+ , "--entrypoint"
+ , "/bin/sh"
+ , "docker.io/temporalio/admin-tools:1.25.2-tctl-1.18.1-cli-1.1.1"
+ , "-c"
+ , "temporal operator search-attribute list | grep -qw MirrorName || temporal operator search-attribute create --name MirrorName --type Text --namespace default"
+ ]
+ detached "flow-api" (["-p", "127.0.0.1:" <> Text.unpack apiPort <> ":8113"] <> catalog <> flow <> s3 <> ["ghcr.io/peerdb-io/flow-api:" <> peerdbTag])
+ detached "flow-worker" (catalog <> flow <> s3 <> ["ghcr.io/peerdb-io/flow-worker:" <> peerdbTag])
+ detached "flow-snapshot-worker" (catalog <> flow <> s3 <> ["ghcr.io/peerdb-io/flow-snapshot-worker:" <> peerdbTag])
+ -- the source: PostgreSQL's own default wal_level (replica), so the node has something to do
+ detached "source" ["-e", "POSTGRES_PASSWORD=unused-admin", "docker.io/library/postgres:17-alpine"]
+ let adminXml = dir "salmon-admin.xml"
+ writeFile adminXml (Text.unpack ClickHouse.renderAdminAccess)
+ detached "ch" ["-v", adminXml <> ":/etc/clickhouse-server/users.d/salmon-admin.xml:ro", "docker.io/clickhouse/clickhouse-server:25.8"]
+ waitUntil "the source" 60 (execOk (name "source") ["pg_isready", "-U", "postgres"])
+ waitUntil "ClickHouse" 60 (execOk (name "ch") ["clickhouse-client", "--query", "SELECT 1"])
+ waitUntil "the flow API" 120 (ok "curl" ["-sf", "http://127.0.0.1:" <> Text.unpack apiPort <> "/v1/version"])
+
+converge :: FilePath -> String -> (String -> String) -> Text -> IO ()
+converge dir net name apiPort = do
+ let src = name "source"
+ ch = name "ch"
+ pgPass = dir "cdc.pass"
+ chPass = dir "peer.pass"
+ api = PeerDB.Api ("http://127.0.0.1:" <> apiPort)
+ chClient = ClickHouse.Client (ClickHouse.InContainer (Podman.ContainerName (Text.pack ch))) ClickHouse.defaultTcpPort
+ writeFile pgPass "cdc-password"
+ writeFile chPass "peer-password"
+ let walLevel = Logical.logicalWalLevel silent ignoreTrack 5432
+ withSudoShim src $ do
+ -- wal_level first, alone: it needs the restart that pg_ctlcluster
+ -- would do on Debian, done here by podman
+ assertEqual "wal_level starts at replica" (Failure "wal_level is replica, logical decoding needs logical") =<< (nodeOf walLevel).check
+ okWal <- runUp walLevel
+ assertBool "wal_level written" okWal
+ assertEqual "the wal_level check reads the files" Success =<< (nodeOf walLevel).check
+ podman ["restart", src]
+ waitUntil "the source after its restart" 60 (execOk src ["pg_isready", "-U", "postgres"])
+ psqlOut src "SHOW wal_level" >>= assertEqual "running wal_level" "logical"
+ -- peers name their hosts by address, taken after the restart (which
+ -- changes it): podman's resolver has been seen to time out for a
+ -- container a few seconds old, and PeerDB refuses a peer it cannot reach
+ srcIp <- addressOf net src
+ chIp <- addressOf net ch
+ psql src "CREATE TABLE orders (id int PRIMARY KEY, item text); INSERT INTO orders VALUES (1, 'apple'), (2, 'pear')"
+ let pgUser = Postgres.userPassFile silent ignoreTrack ignoreTrack 5432 (FS.PreExisting pgPass) (Postgres.User "cdc")
+ role = Logical.replicationRole silent ignoreTrack 5432 "cdc" `inject` pgUser
+ publication = Logical.publication silent ignoreTrack 5432 (Logical.Publication "cdc_pub" "postgres" [Logical.publicTable "orders"])
+ slot = Logical.logicalSlot silent ignoreTrack 5432 "postgres" "cdc_slot" `inject` walLevel
+ chDb = ClickHouse.database silent ignoreTrack chClient "analytics"
+ chUser = ClickHouse.userPassFile silent ignoreTrack chClient (FS.PreExisting chPass) "peer"
+ chGrant = (ClickHouse.grant silent ignoreTrack chClient (ClickHouse.Grant ["SELECT", "INSERT", "CREATE TABLE", "ALTER", "DROP TABLE", "TRUNCATE", "OPTIMIZE"] "analytics" "peer") `inject` chUser) `inject` chDb
+ chS3 = ClickHouse.sourceReadGrant silent ignoreTrack chClient "S3" "peer" `inject` chUser
+ source = PeerDB.Peer "pg_source" (PeerDB.PostgresSource (PeerDB.PostgresPeer srcIp 5432 "cdc" "postgres" False))
+ target = PeerDB.Peer "ch_target" (PeerDB.ClickHouseTarget (PeerDB.ClickHousePeer chIp 9000 "peer" "analytics" True))
+ srcPeerOp = (PeerDB.peer silent ignoreTrack api (FS.PreExisting pgPass) source `inject` role) `inject` walLevel
+ dstPeerOp = (PeerDB.peer silent ignoreTrack api (FS.PreExisting chPass) target `inject` chGrant) `inject` chS3
+ -- PeerDB's own slot, and the snapshot it exports with it
+ snapshotting =
+ (PeerDB.mirror "orders_snapshot" "pg_source" "ch_target" [PeerDB.TableMapping "public.orders" "orders"])
+ { PeerDB.mirrorPublication = Just "cdc_pub"
+ , PeerDB.mirrorIdleTimeoutSeconds = Just 3
+ }
+ -- the declared slot: no snapshot (PeerDB refuses one on an existing slot)
+ declared =
+ (PeerDB.mirror "orders_from_slot" "pg_source" "ch_target" [PeerDB.TableMapping "public.orders" "orders_cdc"])
+ { PeerDB.mirrorPublication = Just "cdc_pub"
+ , PeerDB.mirrorSlot = Just "cdc_slot"
+ , PeerDB.mirrorInitialSnapshot = False
+ , PeerDB.mirrorIdleTimeoutSeconds = Just 3
+ }
+ withPeers m = ((PeerDB.mirrorOp silent ignoreTrack api m `inject` srcPeerOp) `inject` dstPeerOp) `inject` publication
+ snapshotOp = withPeers snapshotting
+ mirrorOp = withPeers declared `inject` slot
+ graph = op "peerdb-test" (deps [snapshotOp, mirrorOp]) id
+ withSudoShim src $ do
+ okRole <- runUp role
+ assertBool "the CDC role, with REPLICATION" okRole
+ -- what the declaration does not hold: the CDC role reads the table, and
+ -- may create PeerDB's metadata schema in the database
+ psql src "GRANT SELECT ON orders TO cdc; GRANT CREATE ON DATABASE postgres TO cdc"
+ withSudoShim src $ do
+ first <- runUpCapturing graph
+ assertBool ("first pass: " <> show (failures first)) (null (failures first))
+ forM_ [("wal_level", walLevel), ("role", role), ("publication", publication), ("slot", slot), ("ch database", chDb), ("ch user", chUser), ("ch grant", chGrant), ("ch s3", chS3), ("source peer", srcPeerOp), ("target peer", dstPeerOp)] $ \(what, o) ->
+ assertEqual what Success =<< (nodeOf o).check
+ forM_ [("orders_snapshot", snapshotOp), ("orders_from_slot", mirrorOp)] $ \(mname, o) ->
+ waitForCheck ("mirror " <> mname) 180 (nodeOf o).check $ do
+ (_, out, _) <-
+ readProcessWithExitCode
+ "curl"
+ ["-s", "-X", "POST", "-d", "{\"flow_job_name\":\"" <> mname <> "\",\"exclude_batches\":true}", "http://127.0.0.1:" <> Text.unpack apiPort <> "/v1/mirrors/status"]
+ ""
+ pure ("PeerDB says: " <> take 3000 out)
+ -- the initial snapshot on one, nothing on the other; then a change streamed to both
+ waitUntil "the snapshot in ClickHouse" 120 ((== "1:apple,2:pear") <$> chRows ch "orders")
+ psql src "INSERT INTO orders VALUES (3, 'plum'); UPDATE orders SET item = 'quince' WHERE id = 2"
+ waitUntil "the change through PeerDB's slot" 120 ((== "1:apple,2:quince,3:plum") <$> chRows ch "orders")
+ waitUntil "the change through the declared slot" 120 ((== "2:quince,3:plum") <$> chRows ch "orders_cdc")
+ psqlOut src "SELECT active FROM pg_replication_slots WHERE slot_name = 'cdc_slot'" >>= assertEqual "PeerDB reads the declared slot" "t"
+ psqlOut src "SELECT string_agg(slot_name, ',' ORDER BY slot_name) FROM pg_replication_slots" >>= assertEqual "and PeerDB made one for the other mirror" "cdc_slot,peerflow_slot_orders_snapshot"
+ -- a declared slot with a snapshot is refused before PeerDB is asked
+ withSudoShim src $ do
+ let bad = declared{PeerDB.mirrorName = "orders_bad", PeerDB.mirrorInitialSnapshot = True}
+ badOp = PeerDB.mirrorOp silent ignoreTrack api bad
+ refusedBad <- runUp badOp
+ assertBool "a declared slot with a snapshot is refused" (not refusedBad)
+ assertEqual "and PeerDB was not asked" (Failure "PeerDB has no mirror orders_bad") =<< (nodeOf badOp).check
+ withSudoShim src $ do
+ second <- runUpCapturing graph
+ -- what runs again is what has no check: the root and the two
+ -- pre-existing password files' placeholders
+ reapplied <- sequence [(\r -> (Text.unpack a.shorthand, r)) <$> a.extension.check | UpDown.Done a <- second]
+ assertEqual "the second pass applies nothing that has a check" [] [x | x@(_, r) <- reapplied, r /= Immaterial]
+ -- a mirror declared with another table: a Failure, and up refuses
+ let other = declared{PeerDB.mirrorTables = [PeerDB.TableMapping "public.orders" "orders", PeerDB.TableMapping "public.other" "other"]}
+ otherOp = PeerDB.mirrorOp silent ignoreTrack api other
+ otherCheck <- (nodeOf otherOp).check
+ assertBool ("another table set is a Failure: " <> show otherCheck) (isFailure otherCheck)
+ refused <- runUp otherOp
+ assertBool "up refuses to change a mirror" (not refused)
+ -- a publication salmon did not make is left alone
+ psql src "CREATE PUBLICATION by_hand FOR TABLE orders"
+ let byHand = Logical.publication silent ignoreTrack 5432 (Logical.Publication "by_hand" "postgres" [Logical.publicTable "orders"])
+ assertEqual "unmarked" (Failure "publication by_hand exists but salmon did not make it") =<< (nodeOf byHand).check
+ refusedPub <- runUp byHand
+ assertBool "up refuses an unmarked publication" (not refusedPub)
+ -- down takes nothing away
+ downOk <- runDown graph
+ assertBool "down succeeds" downOk
+ assertEqual "the mirrors still run" [Success, Success] =<< mapM (\o -> (nodeOf o).check) [snapshotOp, mirrorOp]
+ psqlOut src "SELECT count(*) FROM pg_publication WHERE pubname IN ('cdc_pub', 'by_hand')" >>= assertEqual "both publications remain" "2"
+ psqlOut src "SELECT obj_description(oid, 'pg_publication') IS NULL FROM pg_publication WHERE pubname = 'by_hand'" >>= assertEqual "the hand-made one is unmarked still" "t"
+ where
+ failures :: [UpDown.Report Extension] -> [String]
+ failures reports = [Text.unpack a.extension.help <> ": " <> show e | UpDown.Failed a e <- reports]
+
+-------------------------------------------------------------------------------
+
+teardown :: String -> (String -> String) -> IO ()
+teardown net name = do
+ forM_ ["flow-snapshot-worker", "flow-worker", "flow-api", "temporal", "minio", "catalog", "source", "ch"] $ \n ->
+ void (readProcessWithExitCode "podman" ["rm", "-f", "-t", "2", name n] "")
+ void (readProcessWithExitCode "podman" ["network", "rm", net] "")
+
+podman :: [String] -> IO ()
+podman args = do
+ (code, out, err) <- readProcessWithExitCode "podman" args ""
+ unless (code == ExitSuccess) $ assertFailure ("podman " <> unwords (take 6 args) <> " failed:\n" <> out <> err)
+
+ok :: String -> [String] -> IO Bool
+ok exe args = do
+ r <- try (readProcessWithExitCode exe args "") :: IO (Either SomeException (ExitCode, String, String))
+ pure (either (const False) (\(c, _, _) -> c == ExitSuccess) r)
+
+execOk :: String -> [String] -> IO Bool
+execOk cid args = ok "podman" (["exec", cid] <> args)
+
+waitUntil :: String -> Int -> IO Bool -> IO ()
+waitUntil what seconds probe = go seconds
+ where
+ go n = do
+ done <- probe
+ unless done $ do
+ when (n <= 0) $ assertFailure (what <> " not ready in " <> show seconds <> "s")
+ threadDelay 1000000
+ go (n - 1)
+
+psql :: String -> String -> IO ()
+psql cid sql = podman ["exec", "-u", "postgres", cid, "psql", "-v", "ON_ERROR_STOP=1", "-c", sql]
+
+-- | A container's address on a network.
+addressOf :: String -> String -> IO Text
+addressOf net cid = do
+ (code, out, err) <- readProcessWithExitCode "podman" ["inspect", "-f", "{{(index .NetworkSettings.Networks \"" <> net <> "\").IPAddress}}", cid] ""
+ unless (code == ExitSuccess && not (null (words out))) $ assertFailure ("no address for " <> cid <> ": " <> err)
+ pure (Text.strip (Text.pack out))
+
+-- | @id:item@ rows of a mirrored table, deduplicated (@FINAL@), deleted rows left out.
+chRows :: String -> String -> IO String
+chRows cid table = do
+ (_, out, _) <-
+ readProcessWithExitCode
+ "podman"
+ ["exec", cid, "clickhouse-client", "--query", "SELECT arrayStringConcat(groupArray(concat(toString(id), ':', item)), ',') FROM (SELECT id, item FROM analytics." <> table <> " FINAL WHERE _peerdb_is_deleted = 0 ORDER BY id)"]
+ ""
+ pure (concat (lines out))
+
+psqlOut :: String -> String -> IO String
+psqlOut cid sql = do
+ (_, out, _) <- readProcessWithExitCode "podman" ["exec", "-u", "postgres", cid, "psql", "-tAc", sql] ""
+ pure (concat (lines out))
+
+{- | @sudo -u USER CMD ...@ as @podman exec -i -u USER CONTAINER CMD ...@: the
+Postgres nodes run @sudo -u postgres psql@, and the official image has no
+@sudo@.
+-}
+withSudoShim :: String -> IO a -> IO a
+withSudoShim cid act =
+ withTempDir $ \dir -> do
+ writeExecutable (dir "sudo") $
+ unlines
+ [ "#!/bin/bash"
+ , "u=root"
+ , "if [ \"$1\" = \"-u\" ]; then u=\"$2\"; shift 2; fi"
+ , "exec podman exec -i -u \"$u\" " <> cid <> " \"$@\""
+ ]
+ withPrependedPath dir act
+
+-- | Waits for a check to read 'Success', and says what it read last when it does not.
+waitForCheck :: String -> Int -> IO CheckResult -> IO String -> IO ()
+waitForCheck what seconds check diagnose = go seconds
+ where
+ go n = do
+ r <- check
+ unless (r == Success) $ do
+ when (n <= 0) $ do
+ d <- diagnose
+ assertFailure (what <> " not Success in " <> show seconds <> "s, last read " <> show r <> "\n" <> d)
+ threadDelay 1000000
+ go (n - 1)
diff --git a/test/Test/ServeJsonLogSpec.hs b/test/Test/ServeJsonLogSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Test/ServeJsonLogSpec.hs
@@ -0,0 +1,67 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+{- | Layer 0 coverage for @run serve --json-log FILE@: what the flag parses
+into ('CommandLine.ReportTextJsonLog'), that it excludes @--json@, and that
+@run up@\/@run down@ do not take it. What the loop then does with it (the
+text reporters on stdout, 'Salmon.Reporter.Tagged.reportJSONLines' on the
+file beside them) is composed in 'CommandLine.execCommandOrSeedWithRewrites',
+which reads the process's own stdin and so is not run in process here.
+-}
+module Test.ServeJsonLogSpec (tests) where
+
+import Data.Aeson (Result (..), Value (..), fromJSON, toJSON)
+import qualified Data.Aeson.KeyMap as KeyMap
+import Data.Foldable (toList)
+import qualified Options.Applicative as Opt
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertEqual, assertFailure, testCase)
+
+import qualified Salmon.Builtin.CommandLine as CommandLine
+
+tests :: TestTree
+tests =
+ testGroup
+ "serve --json-log"
+ [ testCase "--json-log FILE is text on the terminal and JSON in FILE" $ do
+ fmt <- serveFormat ["serve", "--json-log", "/var/log/salmon.jsonl"]
+ assertEqual "" (CommandLine.ReportTextJsonLog "/var/log/salmon.jsonl") fmt
+ , testCase "without a flag, and with --json, nothing changed" $ do
+ serveFormat ["serve"] >>= assertEqual "text" CommandLine.ReportText
+ serveFormat ["serve", "--json"] >>= assertEqual "json" CommandLine.ReportJson
+ , testCase "--json and --json-log together are refused" $ do
+ refused ["serve", "--json", "--json-log", "/tmp/x.jsonl"]
+ refused ["serve", "--json-log", "/tmp/x.jsonl", "--json"]
+ , testCase "run up and run down do not take --json-log" $ do
+ refused ["up", "--json-log", "/tmp/x.jsonl"]
+ refused ["down", "--json-log", "/tmp/x.jsonl"]
+ ]
+
+parse :: [String] -> Opt.ParserResult CommandLine.RunCommand
+parse = Opt.execParserPure Opt.defaultPrefs (Opt.info CommandLine.runCommandParser mempty)
+
+refused :: [String] -> IO ()
+refused args =
+ case parse args of
+ Opt.Failure _ -> pure ()
+ Opt.Success other -> assertFailure (unwords args <> ": accepted as " <> show other)
+ Opt.CompletionInvoked _ -> assertFailure "completion"
+
+{- | The 'CommandLine.ReportFormat' of a parsed @serve@, read as the fourth
+of the constructor's fields (after the concurrency cap, the
+'CommandLine.StartupOptions' and @--no-autoconverge@) through its JSON encoding rather than by a
+pattern naming all of them, so a flag appended to 'CommandLine.RunServe'
+does not have to be spelled here too. -}
+serveFormat :: [String] -> IO CommandLine.ReportFormat
+serveFormat args =
+ case parse args of
+ Opt.Success cmd@CommandLine.RunServe{} ->
+ case toJSON cmd of
+ Object o
+ | Just (Array fields) <- KeyMap.lookup "contents" o
+ , (_ : _ : _ : v : _) <- toList fields
+ , Success fmt <- fromJSON v ->
+ pure fmt
+ other -> assertFailure ("unexpected encoding of a serve command: " <> show other)
+ Opt.Success other -> assertFailure ("not a serve command: " <> show other)
+ Opt.Failure f -> assertFailure ("parse failed: " <> fst (Opt.renderFailure f "salmon"))
+ Opt.CompletionInvoked _ -> assertFailure "completion"
diff --git a/test/Test/ServeSpec.hs b/test/Test/ServeSpec.hs
--- a/test/Test/ServeSpec.hs
+++ b/test/Test/ServeSpec.hs
@@ -21,6 +21,7 @@
import Data.Aeson (FromJSON, ToJSON, encode)
import qualified Data.ByteString.Lazy as LByteString
import Data.Dynamic (toDyn)
+import Data.Functor.Contravariant (contramap)
import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
import qualified Data.List
import qualified Data.Set as Set
@@ -37,12 +38,15 @@
import System.Timeout (timeout)
import Test.Tasty (TestTree, testGroup)
import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)
+import qualified Options.Applicative as Opt
import qualified Salmon.Actions.Serve as Serve
import Salmon.Actions.Serve (Convergence (..), Direction (..), NodeState (..), World (..))
import qualified Salmon.Actions.UpDown as UpDown
import Salmon.Actions.UpDown (CheckResult (..))
import qualified Salmon.Actions.Upkeep as Upkeep
+import qualified Salmon.Actions.Follow as Follow
+import qualified Salmon.Builtin.CommandLine as CommandLine
import Salmon.Builtin.Extension (Extension, Op, Track', check, deps, down, dynamics, managed, nodeps, notes, op, opAct, ref, up)
import qualified Salmon.Builtin.Nodes.Daemon as Daemon
import qualified Salmon.Builtin.Nodes.Filesystem as FS
@@ -56,7 +60,8 @@
import Salmon.Op.Supervision (Strategy (..), Supervision (..), defaultSupervision, supervised)
import qualified Salmon.Op.Rewrite as Rewrite
import Salmon.Op.Track (Track (..))
-import Salmon.Reporter (Reporter, ReporterM (..))
+import qualified Salmon.Op.Window as Window
+import Salmon.Reporter (Reporter, ReporterM (..), silent)
import Test.Harness (capture, privatePipe, withTempDir)
@@ -78,6 +83,9 @@
, testCase "unparseable input does not disturb the world" badInputIsInert
, testCase "`load` runs a file of declarations as if typed" loadRunsAScript
, testCase "`up-directive` declares straight from a directive file" upDirectiveDeclares
+ , testCase "quoteArg spells any file name as one line-protocol word" quoteArgRoundTrips
+ , testCase "--no-supervise and --up-directive parse into the start-up lines, and default to none" startupFlagsParse
+ , testCase "start-up lines run before stdin is read, and declare under their own origin" startupLinesRunFirst
, testCase "`status --exclude **` hides every node" statusExcludeAllHidesEverything
, testCase "`history --exclude **` hides every epoch" historyExcludeAllHidesEverything
, testCase "`converge --select` restricted to nothing leaves a failed node untouched" convergeSelectRestricts
@@ -105,6 +113,12 @@
, testCase "`status` prints each node's path, and it round-trips as a `--select` pattern" statusPathsRoundTripAsSelectors
, testCase "two nodes sharing one shorthand-derived path are told apart by `#ref` instead" ambiguousPathsAreDisambiguatedByRef
, testCase "`force` still reaches a node deferred by `autoconverge off`, without converging anything else" autoConvergeOffForceStillReachesANamedNode
+ , testCase "--once: the loop ends after the first pass, leaving queued input unread" onceEndsAfterFirstPass
+ , testCase "--once: a failed first pass is the verdict, and nothing retries it" onceReportsAFailedPass
+ , testCase "--once: commands before the pass run, the explicit `converge` is the pass" onceWaitsForTheFirstPass
+ , testCase "--once: a loaded file stops at the line whose pass it was" onceStopsALoadedFile
+ , testCase "--once: input that ends before any pass ends the loop with no verdict" onceWithoutAPass
+ , testCase "--once --up-directive: the directive's pass is the one pass, stdin and a second directive are not read" onceWithUpDirective
]
-------------------------------------------------------------------------------
@@ -440,6 +454,82 @@
[["", directivePath]]
(fmap Serve.logTokens w.worldLog)
+quoteArgRoundTrips :: IO ()
+quoteArgRoundTrips =
+ mapM_
+ ( \name ->
+ assertEqual
+ ("round trip of " <> show name)
+ (Right ["up-directive", name])
+ (Serve.tokenize ("up-directive " <> Serve.quoteArg name))
+ )
+ ["plain.json", "with space.json", "it's.json", "back\\slash \"dq\".json", "", "''"]
+
+startupFlagsParse :: IO ()
+startupFlagsParse = do
+ none <- parsed ["serve"]
+ assertEqual "no flag, nothing run first" CommandLine.noStartup none
+ assertEqual "no flag, no line" [] (CommandLine.startupLines none)
+ both <- parsed ["serve", "--up-directive", "b c.json", "--no-supervise", "--up-directive", "a.json"]
+ assertEqual "parsed" (CommandLine.StartupOptions False ["b c.json", "a.json"]) both
+ assertEqual
+ "supervise off first, then each directive in order"
+ ["supervise off", "up-directive 'b c.json'", "up-directive 'a.json'"]
+ (CommandLine.startupLines both)
+ where
+ parsed args =
+ case Opt.execParserPure Opt.defaultPrefs (Opt.info CommandLine.runCommandParser mempty) args of
+ Opt.Success (CommandLine.RunServe _ startup _ _ _ _ _ _ _ _ _ _ _ _ _ _ _) -> pure startup
+ Opt.Success other -> assertFailure ("not a serve command: " <> show other)
+ Opt.Failure f -> assertFailure ("parse failed: " <> fst (Opt.renderFailure f "salmon"))
+ Opt.CompletionInvoked _ -> assertFailure "completion"
+
+{- | What @run serve --no-supervise --up-directive FILE@ wires up: the
+start-up producer, and stdin gated on it. Stdin's own lines are already
+there to read when the loop starts, so the order is the gate's doing. -}
+startupLinesRunFirst :: IO ()
+startupLinesRunFirst =
+ withTempDir $ \root -> do
+ let directivePath = root "it's a directive.json"
+ LByteString.writeFile directivePath (encode (Spec (root "files") ["a"]))
+ let opts = CommandLine.StartupOptions False [directivePath]
+ (serveReporter, readServeReports) <- capture
+ (nodeReporter, _) <- capture
+ gate <- newEmptyMVar
+ w <-
+ withScript ["up b"] $ \h ->
+ Serve.serveProducers
+ []
+ Nothing
+ True
+ serveReporter
+ nodeReporter
+ (parseSpec root)
+ (Configure pure)
+ program
+ [ Serve.linesProducer (Serve.Origin "startup") (CommandLine.startupLines opts) (putMVar gate ())
+ , Follow.gated gate (Serve.stdinProducer h)
+ ]
+ reports <- readServeReports
+ assertFileExists root "a" True
+ assertFileExists root "b" True
+ assertAllConverged TurnUp w
+ assertEqual
+ "the directive, then stdin's line"
+ [(Serve.Origin "startup", ["", directivePath]), (Serve.Stdin, ["b"])]
+ (reverse [(e.logOrigin, e.logTokens) | e <- w.worldLog])
+ assertEqual "supervision was turned off, once" [False] [on | Serve.Supervised on <- reports]
+ assertBool
+ "and before the first pass"
+ (case dropWhile (not . isSupervised) reports of
+ (_ : rest) -> any isConvergeStart rest
+ [] -> False)
+ where
+ isSupervised Serve.Supervised{} = True
+ isSupervised _ = False
+ isConvergeStart Serve.ConvergeStart{} = True
+ isConvergeStart _ = False
+
statusExcludeAllHidesEverything :: IO ()
statusExcludeAllHidesEverything =
withTempDir $ \root -> do
@@ -565,6 +655,34 @@
Serve.serveWith rewrites Nothing True serveReporter nodeReporter (parseSpec root) (Configure pure) prog
(,,) w <$> readServeReports <*> readNodeReports
+-- | 'runServe' under 'Serve.AfterFirstPass', the loop @run serve --once@ runs.
+runServeOnce ::
+ Track' Spec ->
+ FilePath ->
+ [String] ->
+ IO (World Spec Spec, [Serve.Report], [UpDown.Report Extension])
+runServeOnce prog root script = do
+ (serveReporter, readServeReports) <- capture
+ (nodeReporter, readNodeReports) <- capture
+ w <-
+ withScript script $ \h ->
+ Serve.serveSayingEnding
+ Serve.AfterFirstPass
+ Window.noWindows
+ silent
+ (const (pure ()))
+ []
+ Nothing
+ True
+ (contramap Serve.attributed serveReporter)
+ (contramap Serve.attributed nodeReporter)
+ (parseSpec root)
+ (Configure pure)
+ prog
+ Nothing
+ [Serve.stdinProducer h]
+ (,,) w <$> readServeReports <*> readNodeReports
+
withScript :: [String] -> (Handle -> IO a) -> IO a
withScript ls act =
withSystemTempFile "salmon-serve-script" $ \path h -> do
@@ -1424,3 +1542,109 @@
[(0, 2), (0, 1)]
(convergeStarts reports)
assertAllConverged TurnUp w
+
+-------------------------------------------------------------------------------
+-- --once ('Serve.AfterFirstPass')
+
+{- | Every line of a piped script is in the inbox before the first pass ends,
+so this is the case that tells "end after the pass" from "quit once input is
+done": the second declaration must not be read, let alone converged. -}
+onceEndsAfterFirstPass :: IO ()
+onceEndsAfterFirstPass =
+ withTempDir $ \root -> do
+ (w, reports, _) <- runServeOnce program root ["up a", "up b", "status"]
+ assertFileExists root "a" True
+ assertFileExists root "b" False
+ assertAllConverged TurnUp w
+ assertEqual "exactly one pass, and it converged" [(True, 0)] (convergeStops reports)
+ assertEqual "one declaration read" 1 (length w.worldLog)
+ assertEqual "the queued `status` was not read" [] [() | Serve.StatusReport{} <- reports]
+ assertEqual "input did not close, so no `Stopped`" [] [() | Serve.Stopped <- reports]
+
+onceReportsAFailedPass :: IO ()
+onceReportsAFailedPass =
+ withTempDir $ \root -> do
+ attempts <- newIORef (0 :: Int)
+ (_, reports, _) <- runServeOnce (flaky attempts) root ["up a", "converge"]
+ assertEqual "attempted once: the second pass never ran" 1 =<< readIORef attempts
+ assertEqual "the verdict is the failed pass's" [(False, 1)] (convergeStops reports)
+
+{- | The pass, not the first command, is what ends the loop: settings and
+declarations recorded under @autoconverge off@ run, and the explicit
+@converge@ is the one pass. -}
+onceWaitsForTheFirstPass :: IO ()
+onceWaitsForTheFirstPass =
+ withTempDir $ \root -> do
+ (_, reports, _) <- runServeOnce program root ["supervise off", "autoconverge off", "up a", "up b", "converge", "up c"]
+ assertFileExists root "a" True
+ assertFileExists root "b" True
+ assertFileExists root "c" False
+ assertEqual "one pass" [(True, 0)] (convergeStops reports)
+
+onceStopsALoadedFile :: IO ()
+onceStopsALoadedFile =
+ withTempDir $ \root -> do
+ let scriptPath = root "commands.txt"
+ writeFile scriptPath (unlines ["up a", "up b"])
+ (_, reports, _) <- runServeOnce program root ["load " <> scriptPath, "up c"]
+ assertFileExists root "a" True
+ assertFileExists root "b" False
+ assertFileExists root "c" False
+ assertEqual "one pass" [(True, 0)] (convergeStops reports)
+ assertEqual "the file was not run to its end" [] [() | Serve.LoadDone{} <- reports]
+
+onceWithoutAPass :: IO ()
+onceWithoutAPass =
+ withTempDir $ \root -> do
+ (w, reports, _) <- runServeOnce program root ["autoconverge off", "up a"]
+ assertFileExists root "a" False
+ assertEqual "no pass ran" [] (convergeStops reports)
+ assertEqual "the declaration was recorded" 1 (length w.worldLog)
+ assertEqual "the loop ended on end of input" [()] [() | Serve.Stopped <- reports]
+
+{- | What @run serve --once --up-directive F1 --up-directive F2@ wires up:
+the start-up lines, then stdin gated on them. The first directive's
+auto-converge is the one pass, so neither the second directive nor stdin's
+line is read; @supervise off@ from the start-up lines is what @--once@
+already starts with. -}
+onceWithUpDirective :: IO ()
+onceWithUpDirective =
+ withTempDir $ \root -> do
+ let first = root "first.json"
+ second = root "second.json"
+ LByteString.writeFile first (encode (Spec (root "files") ["a"]))
+ LByteString.writeFile second (encode (Spec (root "files") ["b"]))
+ let opts = CommandLine.StartupOptions True [first, second]
+ (serveReporter, readServeReports) <- capture
+ gate <- newEmptyMVar
+ w <-
+ withScript ["up c"] $ \h ->
+ Serve.serveSayingEnding
+ Serve.AfterFirstPass
+ Window.noWindows
+ silent
+ (const (pure ()))
+ []
+ Nothing
+ True
+ (contramap Serve.attributed serveReporter)
+ silent
+ (parseSpec root)
+ (Configure pure)
+ program
+ Nothing
+ [ Serve.linesProducer (Serve.Origin "startup") (CommandLine.startupLines opts) (putMVar gate ())
+ , Follow.gated gate (Serve.stdinProducer h)
+ ]
+ reports <- readServeReports
+ assertFileExists root "a" True
+ assertFileExists root "b" False
+ assertFileExists root "c" False
+ assertEqual "one pass, and it converged" [(True, 0)] (convergeStops reports)
+ assertEqual
+ "only the first directive was read"
+ [(Serve.Origin "startup", ["", first])]
+ [(e.logOrigin, e.logTokens) | e <- w.worldLog]
+
+convergeStops :: [Serve.Report] -> [(Bool, Int)]
+convergeStops reports = [(ok, n) | Serve.ConvergeStop ok n <- reports]
diff --git a/test/Test/StatusSinkSpec.hs b/test/Test/StatusSinkSpec.hs
--- a/test/Test/StatusSinkSpec.hs
+++ b/test/Test/StatusSinkSpec.hs
@@ -89,7 +89,7 @@
where
parsed args =
case Opt.execParserPure Opt.defaultPrefs (Opt.info CommandLine.runCommandParser mempty) args of
- Opt.Success (CommandLine.RunServe _ _ _ _ _ _ _ _ _ sink _ _ _ _) -> pure sink
+ Opt.Success (CommandLine.RunServe _ _ _ _ _ _ _ _ _ _ sink _ _ _ _ _ _) -> pure sink
Opt.Success other -> assertFailure ("not a serve command: " <> show other)
Opt.Failure f -> assertFailure ("parse failed: " <> fst (Opt.renderFailure f "salmon"))
Opt.CompletionInvoked _ -> assertFailure "completion"