salmon-ops 0.2.0.0 → 0.3.0.0

raw patch · 7 files changed

+1927/−78 lines, 7 filesdep ~aesondep ~arraydep ~asyncPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: aeson, array, async, base64-bytestring, bytestring, containers, contravariant, cryptohash-sha256, crypton, crypton-connection, crypton-x509-store, directory, file-embed, filepath, free, hashable, http-client, http-client-tls, http-types, jose, mtl, network, optparse-applicative, optparse-generic, process, process-extras, salmon-ops, stm, text, time, tls, unix, wai, warp, warp-tls

API changes (from Hackage documentation)

Files

• CHANGELOG.md +36/−1
• salmon-ops.cabal +62/−59
• src/Salmon/Actions/Serve.hs +103/−8
• src/Salmon/Builtin/CommandLine.hs +167/−10
• src/Salmon/Builtin/Nodes/ClickHouse.hs +649/−0
• src/Salmon/Builtin/Nodes/PeerDB.hs +553/−0
• src/Salmon/Builtin/Nodes/Postgres/Logical.hs +357/−0

@@ -1,6 +1,41 @@ # Revision history for salmon-ops -## 0.2.0.0 -- unreleased+## 0.3.0.0 -- 2026-10-31++Breaking: `Salmon.Builtin.CommandLine`'s `RunServe` constructor gained fields (start-up+options, `--once`, `--linger`) and `ReportFormat` gained a constructor (`ReportTextJsonLog`).+Requires `salmon-core ^>=0.1.0.0` (unchanged).++Packaging: every external dependency now carries a PVP upper bound (`^>=` the version built on+GHC 9.10.3; `base` keeps its range).++`run serve`:++* `--once` ends the loop after the first convergence pass and exits with its verdict (0 only+  when the pass converged with nothing left); it starts with supervision off. `--linger SECONDS`+  keeps `--listen`/`--http`/`--http-tcp` answering reads from the final world that long after.+  In the library: `Serve.Ending` (`UntilQuit` | `AfterFirstPass`) and `serveSayingEnding`;+  existing entry points keep `UntilQuit`.+* `--json-log FILE`: the text reports on the terminal as before, plus the objects `--json`+  would print appended to FILE, one per line. Excludes `--json`.+* `--no-supervise` and `--up-directive FILE` run `supervise off` and `up-directive FILE` as+  the loop's first commands, ahead of any other input, so a loop started by a unit file needs+  nothing typed. `Serve.linesProducer` and `Serve.quoteArg` are exported for the same purpose.++New builtins:++* `ClickHouse`: the official image as a system-scope quadlet with watched `config.d`/`users.d`+  drop-ins; `CREATE DATABASE IF NOT EXISTS`, users whose password is a pre-provisioned file,+  database-level grants checked on `system.grants`, and `sourceReadGrant` (`GRANT READ ON S3`)+  for PeerDB. Nothing drops or revokes.+* `Postgres.Logical`: the prerequisites of logical replication (`wal_level = logical`, the+  `REPLICATION` attribute, a publication of exactly the declared tables, a `pgoutput` slot),+  each with a check and no down.+* `PeerDB`: Postgres and ClickHouse peers and CDC mirrors through PeerDB's flow API. A mirror+  that exists with other peers or tables, or has failed, is refused, never changed or recreated;+  nothing is dropped.++## 0.2.0.0 -- 2026-10-23  Breaking: several exported records gained or changed fields (e.g. `Gcp.Compute`'s instance, `Client.Model`'s model) and some exported functions changed signature. Requires

@@ -1,6 +1,6 @@ cabal-version:      2.4 name:               salmon-ops-version:            0.2.0.0+version:            0.3.0.0 synopsis:           IO builtins and command line plumbing for Salmon operations. description:        Builtin operations (files, systemd, debian packages, podman, postgres, wireguard, certificates, ssh, ...) built on salmon-core, plus the shared command line, the long-running serve loop with its HTTP API and web UI. homepage:           https://lucasdicioccio.github.io/salmon/@@ -123,6 +123,9 @@                    , Salmon.Builtin.Nodes.Patroni.RollingMaintenance                    , Salmon.Builtin.Nodes.PgBackRest                    , Salmon.Builtin.Nodes.Postgres+                   , Salmon.Builtin.Nodes.Postgres.Logical+                   , Salmon.Builtin.Nodes.PeerDB+                   , Salmon.Builtin.Nodes.ClickHouse                    , Salmon.Builtin.Nodes.ClipCpp                    , Salmon.Builtin.Nodes.LlamaServer                    , Salmon.Builtin.Nodes.PgVector@@ -162,44 +165,44 @@                    , Salmon.Reporter.Tagged                    , Salmon.Reporter.UpDownText     build-depends:    base >=4.16.3.0 && <4.22-                    , array-                    , async-                    , aeson-                    , base64-bytestring-                    , bytestring-                    , containers-                    , contravariant-                    , crypton-                    , crypton-connection-                    , crypton-x509-store-                    , cryptohash-sha256-                    , directory-                    , file-embed-                    , filepath-                    , free-                    , hashable-                    , http-client-                    , http-client-tls-                    , http-types-                    , jose-                    , mtl-                    , network-                    , optparse-applicative-                    , optparse-generic-                    , process-                    , process-extras+                    , array ^>=0.5.8.0+                    , async ^>=2.2.6+                    , aeson ^>=2.3.2.0+                    , base64-bytestring ^>=1.2.1.0+                    , bytestring ^>=0.12.2.0+                    , containers ^>=0.7+                    , contravariant ^>=1.5.6+                    , crypton ^>=2.2.5+                    , crypton-connection ^>=0.4.6+                    , crypton-x509-store ^>=1.10.0+                    , cryptohash-sha256 ^>=0.11.102.1+                    , directory ^>=1.3.8.5+                    , file-embed ^>=0.0.16.0+                    , filepath ^>=1.5.4.0+                    , free ^>=5.2+                    , hashable ^>=1.5.1.0+                    , http-client ^>=0.7.19+                    , http-client-tls ^>=0.4.0+                    , http-types ^>=0.12.6+                    , jose ^>=0.13+                    , mtl ^>=2.3.1+                    , network ^>=3.2.9.0+                    , 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-                    , stm-                    , text-                    , time-                    , unix-                    , tls+                    , stm ^>=2.5.3.1+                    , text ^>=2.1.3+                    , time ^>=1.12.2+                    , unix ^>=2.8.7.0+                    , tls ^>=2.4.12     hs-source-dirs:   src     if flag(http-server)         exposed-modules: Salmon.Actions.Serve.Http-        build-depends:   wai-                       , warp-                       , warp-tls+        build-depends:   wai ^>=3.2.5+                       , warp ^>=3.4.16+                       , warp-tls ^>=3.4.14         cpp-options:     -DSALMON_HTTP_SERVER     else         other-modules:   Salmon.Actions.Serve.NoHttp@@ -217,10 +220,10 @@     main-is:          DotFixture.hs     hs-source-dirs:   fixtures     build-depends:    base >=4.16.3.0 && <4.22-                    , mtl+                    , mtl ^>=2.3.1                     , salmon-core ^>=0.1.0.0-                    , salmon-ops ^>=0.2.0.0-                    , text+                    , salmon-ops ^>=0.3.0.0+                    , text ^>=2.1.3     -- every salmon binary can end up running the concurrent driver (`run     -- serve` does), and a node's own thread must not block the whole runtime.     ghc-options:      -threaded@@ -233,10 +236,10 @@     main-is:          PostgresReplicationFixture.hs     hs-source-dirs:   fixtures     build-depends:    base >=4.16.3.0 && <4.22-                    , mtl+                    , mtl ^>=2.3.1                     , salmon-core ^>=0.1.0.0-                    , salmon-ops ^>=0.2.0.0-                    , text+                    , salmon-ops ^>=0.3.0.0+                    , text ^>=2.1.3     -- every salmon binary can end up running the concurrent driver (`run     -- serve` does), and a node's own thread must not block the whole runtime.     ghc-options:      -threaded@@ -249,12 +252,12 @@     main-is:          QemuHostSetupFixture.hs     hs-source-dirs:   fixtures     build-depends:    base >=4.16.3.0 && <4.22-                    , directory-                    , mtl-                    , filepath+                    , directory ^>=1.3.8.5+                    , mtl ^>=2.3.1+                    , filepath ^>=1.5.4.0                     , salmon-core ^>=0.1.0.0-                    , salmon-ops ^>=0.2.0.0-                    , text+                    , salmon-ops ^>=0.3.0.0+                    , text ^>=2.1.3     -- every salmon binary can end up running the concurrent driver (`run     -- serve` does), and a node's own thread must not block the whole runtime.     ghc-options:      -threaded@@ -267,12 +270,12 @@     main-is:          QemuSmokeRootfsFixture.hs     hs-source-dirs:   fixtures     build-depends:    base >=4.16.3.0 && <4.22-                    , directory-                    , mtl-                    , filepath+                    , directory ^>=1.3.8.5+                    , mtl ^>=2.3.1+                    , filepath ^>=1.5.4.0                     , salmon-core ^>=0.1.0.0-                    , salmon-ops ^>=0.2.0.0-                    , text+                    , salmon-ops ^>=0.3.0.0+                    , text ^>=2.1.3     ghc-options:      -threaded     default-language: Haskell2010     default-extensions: OverloadedStrings@@ -283,15 +286,15 @@     main-is:          ServeFixture.hs     hs-source-dirs:   fixtures     build-depends:    base >=4.16.3.0 && <4.22-                    , aeson-                    , directory-                    , filepath-                    , optparse-applicative-                    , optparse-generic-                    , process+                    , aeson ^>=2.3.2.0+                    , directory ^>=1.3.8.5+                    , filepath ^>=1.5.4.0+                    , optparse-applicative ^>=0.19.0.0+                    , optparse-generic ^>=1.5.3+                    , process ^>=1.6.26.1                     , salmon-core ^>=0.1.0.0-                    , salmon-ops ^>=0.2.0.0-                    , text+                    , salmon-ops ^>=0.3.0.0+                    , text ^>=2.1.3     -- every salmon binary can end up running the concurrent driver (`run     -- serve` does), and a node's own thread must not block the whole runtime.     ghc-options:      -threaded

@@ -132,6 +132,8 @@     serveObservedWith,     serveSaying,     serveSayingWith,+    serveSayingEnding,+    Ending (..),     Followed (..),     AppliedDocument (..),     Mode (..),@@ -146,6 +148,8 @@     originName,     handleProducer,     stdinProducer,+    linesProducer,+    quoteArg,      -- * Attributing reports     Attributed (..),@@ -432,6 +436,9 @@     -- started and adopted), then clears the queue. "Force this node next     -- time you look at it" rather than keeping every one-shot machine alive     -- just so it has a mailbox to post into.+    , tendingPassed :: !(IORef Bool)+    -- ^ set when a convergence pass has reported its 'ConvergeStop': what+    -- 'AfterFirstPass' ends the loop on     }  {- | A representative that lost to the magma's current one, kept for as@@ -1545,6 +1552,34 @@ stdinProducer :: Handle -> Producer stdinProducer = handleProducer Stdin +{- | A fixed list of lines, pushed in order, then @primed@ — what @run serve+--no-supervise@ and @--up-directive FILE@ are: the commands an operator would+otherwise type first, under their own 'Origin'. A caller that wants them run+before anything else gates its other producers on @primed@ (the way the+fetcher's first round gates standard input, see+'Salmon.Actions.Follow.gated'): the inbox is one+queue, so lines pushed before the gate opens are read first.++It pushes no 'Eof': nobody waits on this origin's hang-up (no connection+belongs to it), and an 'Eof' from it would only add a 'HungUp' report.+-}+linesProducer :: Origin -> [String] -> IO () -> Producer+linesProducer origin ls primed = Producer $ \inbox -> do+    atomically (mapM_ (writeTChan inbox . Line origin) ls)+    primed++{- | One argument spelled so that the line protocol's 'tokenize' reads it+back as exactly that one word, spaces, quotes and backslashes included: it+is single-quoted, and a single quote inside it closes the quote, is escaped,+and reopens it. For building a line out of a value the caller holds, such as+a file name given on the command line.+-}+quoteArg :: String -> String+quoteArg a = "'" <> concatMap esc a <> "'"+  where+    esc '\'' = "'\\''"+    esc c = [c]+ {- | 'serveWith', fed by any number of 'Producer's rather than one handle. Each runs on its own thread so that the loop is never itself blocked in a read: the supervisor's machines run while it waits, and stopping them has to@@ -1740,7 +1775,7 @@     IO (World seed directive) serveObservedWith windows observe rewrites limit autoConverge0 rAttributed nodeReporterAttributed parseSeed configure program onFetch producers = do     handling <- newIORef Nothing-    serveLoop windows observe rewrites limit autoConverge0 handling (stamp handling rAttributed) (stamp handling nodeReporterAttributed) parseSeed configure program onFetch producers+    serveLoop UntilQuit windows observe rewrites limit autoConverge0 handling (stamp handling rAttributed) (stamp handling nodeReporterAttributed) parseSeed configure program onFetch producers   where     stamp :: IORef (Maybe Origin) -> Reporter (Attributed a) -> Reporter a     stamp handling = pulls (\rep -> (`Attributed` rep) <$> readIORef handling)@@ -1791,10 +1826,54 @@     Maybe Followed ->     [Producer] ->     IO (World seed directive)-serveSayingWith windows said observe rewrites limit autoConverge0 rAttributed nodeReporterAttributed parseSeed configure program onFetch producers = do+serveSayingWith = serveSayingEnding UntilQuit++{- | When the loop ends of its own accord, besides @quit@ (and 'Eof' from+'Stdin'), which always end it.++'UntilQuit' is every version of 'serve' before the choice existed.+'AfterFirstPass' (@run serve --once@) ends it once the first convergence+pass has run and the command that started it is done — a declaration's+auto-converge, an explicit @converge@, or a fetched 'Batch' alike. Anything+still in the inbox behind that command is left unread, and so are the lines+of a @load@ed file after the one whose pass it was, so exactly one pass runs+however much input was already queued. It also starts with+supervision off (as if @supervise off@ were the first line): a one-pass+session has no "between passes" for a machine to tend in, and a node tended+while declarations wait for an explicit @converge@ would be acted on outside+the pass whose verdict is reported. A @supervise on@ is still obeyed.++The verdict is the first pass's 'ConvergeStop', which the loop reports as it+always does; a caller wanting an exit code reads it off the report stream+(see @run serve --once@ in "Salmon.Builtin.CommandLine"). The loop reports+nothing else on the way out: no 'Stopped', since input did not close.+-}+data Ending = UntilQuit | AfterFirstPass+    deriving (Show, Eq)++-- | 'serveSayingWith', choosing when the loop ends; see 'Ending'.+serveSayingEnding ::+    forall seed directive.+    (ToJSON directive, FromJSON directive) =>+    Ending ->+    Window.Windows ->+    Reporter (Attributed NodeLog.Line) ->+    (IO (World seed directive) -> IO ()) ->+    [Rewrite Extension] ->+    Maybe ConcurrencyLimit ->+    Bool ->+    Reporter (Attributed Report) ->+    Reporter (Attributed (UpDown.Report Extension)) ->+    ([String] -> Either Text seed) ->+    Configure IO seed directive ->+    Track' directive ->+    Maybe Followed ->+    [Producer] ->+    IO (World seed directive)+serveSayingEnding ending windows said observe rewrites limit autoConverge0 rAttributed nodeReporterAttributed parseSeed configure program onFetch producers = do     handling <- newIORef Nothing     NodeLog.withSink (stamp handling said) $-        serveLoop windows observe rewrites limit autoConverge0 handling (stamp handling rAttributed) (stamp handling nodeReporterAttributed) parseSeed configure program onFetch producers+        serveLoop ending windows observe rewrites limit autoConverge0 handling (stamp handling rAttributed) (stamp handling nodeReporterAttributed) parseSeed configure program onFetch producers   where     stamp :: IORef (Maybe Origin) -> Reporter (Attributed a) -> Reporter a     stamp handling = pulls (\rep -> (`Attributed` rep) <$> readIORef handling)@@ -1804,6 +1883,7 @@ serveLoop ::     forall seed directive.     (ToJSON directive, FromJSON directive) =>+    Ending ->     Window.Windows ->     (IO (World seed directive) -> IO ()) ->     [Rewrite Extension] ->@@ -1818,10 +1898,10 @@     Maybe Followed ->     [Producer] ->     IO (World seed directive)-serveLoop windows observe rewrites limit autoConverge0 handling r nodeReporter parseSeed configure program onFetch producers = do+serveLoop ending windows observe rewrites limit autoConverge0 handling r nodeReporter parseSeed configure program onFetch producers = do     world <- newIORef emptyWorld     observe (readIORef world)-    tending <- Tending <$> newIORef Nothing <*> newIORef Upkeep.noKept <*> newIORef True <*> newIORef autoConverge0 <*> newIORef Map.empty+    tending <- Tending <$> newIORef Nothing <*> newIORef Upkeep.noKept <*> newIORef (ending == UntilQuit) <*> newIORef autoConverge0 <*> newIORef Map.empty <*> newIORef False     inbox <- newTChanIO     readers <- traverse (\p -> forkIO (produceInto p inbox)) producers     runReporter r Started@@ -1876,11 +1956,22 @@                         writeIORef handling (Just origin)                         keepGoing <- step tending 0 world origin l                         writeIORef handling Nothing-                        when keepGoing (loop tending world inbox)+                        continue keepGoing                     Batch cmds -> do                         keepGoing <- batch tending world cmds-                        when keepGoing (loop tending world inbox)+                        continue keepGoing+      where+        continue keepGoing = do+            done <- ended tending+            when (keepGoing && not done) (loop tending world inbox) +    -- | Whether 'AfterFirstPass' says the loop is over: checked between+    -- commands, and between the lines of a @load@ed file.+    ended :: Tending -> IO Bool+    ended tending+        | ending == AfterFirstPass = readIORef (tendingPassed tending)+        | otherwise = pure False+     {- | Run a 'Batch': every command with @autoconverge@ held off, the     setting put back afterwards (a @finally@, so a command that stops the     loop still leaves it as the operator had it), then one full convergence@@ -2295,7 +2386,10 @@             pure True         go n (ln : rest) = do             keepGoing <- step tending depth world (Loaded path) ln-            if keepGoing then go (n + 1) rest else pure False+            -- 'AfterFirstPass' stops a file at the line whose pass it was,+            -- unread lines and all (the loop then ends, see 'continue')+            done <- ended tending+            if keepGoing && not done then go (n + 1) rest else pure keepGoing      -- | A 'Configure' that throws is reported as a bad seed and the loop     -- reads on, same as a seed that fails to parse. It used to take the whole@@ -2448,6 +2542,7 @@         w' <- readIORef world         let (rup, rdown) = pendingCounts w'         runReporter r (ConvergeStop (okDown && okUp) (rup + rdown))+        writeIORef (tendingPassed tending) True      {- | Only touch what this pass is for: a node wanted the other way (it     belongs to some other seed), already converged, or excluded by this

@@ -4,7 +4,9 @@  module Salmon.Builtin.CommandLine where +import Control.Concurrent (threadDelay) import Control.Concurrent.MVar (newEmptyMVar, putMVar)+import Data.IORef (modifyIORef', newIORef, readIORef) import Control.Applicative ((<|>)) import Control.Monad (forM, forM_, void, when) import Data.Foldable (traverse_)@@ -14,6 +16,7 @@ import Data.Maybe (fromJust, fromMaybe, isJust) import Data.Set (Set) import qualified Data.Set as Set+import System.Directory (doesFileExist) import Data.Either (rights) import Data.Text (Text) import qualified Data.Text as Text@@ -22,7 +25,8 @@ import qualified Options.Applicative import Options.Generic import System.Exit (exitFailure)-import System.IO (BufferMode (..), hPutStrLn, hSetBuffering, stderr, stdin, stdout)+import Control.Exception (IOException, try)+import System.IO (BufferMode (..), Handle, IOMode (AppendMode), hPutStrLn, hSetBuffering, openFile, stderr, stdin, stdout)  import Salmon.Op.Actions (Act (..)) import qualified Salmon.Op.Concurrency as Concurrency@@ -103,7 +107,9 @@     | -- | @run serve@, optionally capping how many nodes converge at once       -- per pass (R6 in @specs/per-node-state-machines-remaining.md@;       -- 'Nothing' is unbounded, matching every version of @serve@ before-      -- this flag existed), and optionally starting with @autoconverge@ off+      -- this flag existed), what the loop does before reading any input+      -- ('StartupOptions': @--no-supervise@, @--up-directive FILE@), and+      -- optionally starting with @autoconverge@ off       -- (@--no-autoconverge@; 'False' is the default, matching every       -- version of @serve@ before the setting existed). Then pull mode       -- ("Salmon.Actions.Follow"): a registry to follow (@--follow@@ -127,7 +133,11 @@       -- default, is every version before the flag; see 'Http.Access').       -- Then the maintenance windows and @--override-window@, as for       -- 'RunUp' (see 'Serve.serveObservedWith' for what @serve@ does with them).-      RunServe !(Maybe Int) !Bool !ReportFormat !(Maybe FilePath) ![Text] !FollowOptions !(Maybe FilePath) !(Maybe FilePath) !Int !SinkOptions !TcpOptions !Bool ![Text] !Bool+      -- Finally @--once@ ('Serve.AfterFirstPass': end after the first+      -- convergence pass and exit with its verdict) and @--linger SECONDS@+      -- (keep the listeners answering reads that long before exiting; only+      -- with @--once@).+      RunServe !(Maybe Int) !StartupOptions !Bool !ReportFormat !(Maybe FilePath) ![Text] !FollowOptions !(Maybe FilePath) !(Maybe FilePath) !Int !SinkOptions !TcpOptions !Bool ![Text] !Bool !Bool !(Maybe Int)     deriving (Eq, Ord, Generic, Show)  {- | The four flags that put the HTTP surface on a network, as typed. What@@ -277,6 +287,41 @@ instance FromJSON SinkOptions instance ToJSON SinkOptions +{- | What @run serve@ does before it reads a line of input, so that a loop+started by a unit file or a script needs nothing typed on standard input:+@--no-supervise@ is @supervise off@ and each @--up-directive FILE@+(repeatable, in order) is @up-directive FILE@, run as the loop's first+commands under the @startup@ origin and before standard input, a socket+client, an HTTP request or the fetcher's first round is read (see+'Serve.linesProducer'). Neither changes what the commands mean: an+@--up-directive@ converges at once unless @--no-autoconverge@ holds it for a+@converge@, and a @supervise on@ typed later is obeyed. The defaults run+nothing, which is every version before the flags.+-}+data StartupOptions = StartupOptions+    { startupSupervise :: !Bool+    -- ^ 'False' with @--no-supervise@+    , startupDirectives :: ![FilePath]+    -- ^ @--up-directive FILE@, in the order given+    }+    deriving (Eq, Ord, Generic, Show)++instance FromJSON StartupOptions+instance ToJSON StartupOptions++-- | Supervision on, no directive: what @run serve@ starts with by default.+noStartup :: StartupOptions+noStartup = StartupOptions True []++{- | The serve-command lines 'StartupOptions' stands for, in the order they+run: @supervise off@ first, so that nothing a directive brings up is tended+in between, then each directive.+-}+startupLines :: StartupOptions -> [String]+startupLines o =+    ["supervise off" | not o.startupSupervise]+        <> ["up-directive " <> Serve.quoteArg path | path <- o.startupDirectives]+ instance FromJSON RunCommand instance ToJSON RunCommand @@ -340,11 +385,17 @@ binary's own text reporters — so @run up --json | jq@ works, and a client of the server @specs\/generic-server.md@ sketches reads the same objects. Absent, 'ReportText' hands every report to the reporters the binary-passed in, untouched.+passed in, untouched. @run serve --json-log FILE@ selects+'ReportTextJsonLog': the text reports on the terminal exactly as+'ReportText' prints them, and the very objects @--json@ would print+appended to FILE, one per line, flushed per report (so @tail -f FILE | jq@+follows a loop whose operator reads text). It excludes @--json@; @run+up@\/@run down@ do not take it. -} data ReportFormat     = ReportText     | ReportJson+    | ReportTextJsonLog !FilePath     deriving (Eq, Ord, Generic, Show)  instance FromJSON ReportFormat@@ -413,12 +464,13 @@                         <> Options.Applicative.help "Cap how many nodes converge (check/up/down) at once per pass. Omitted = unbounded."                     )                 )+            <*> startupOptionsP             <*> switch                 ( long "no-autoconverge"                     <> Options.Applicative.help                         "Start with `autoconverge off`: declarations are recorded but not converged until an explicit `converge`."                 )-            <*> reportFormatP+            <*> (jsonLogP <|> reportFormatP)             <*> optional                 ( strOption                     ( long "follow"@@ -467,6 +519,37 @@                 )             <*> maintenanceWindowsP "Nodes marked disruptive are brought up only inside these windows (repeatable): outside them a pass skips them without counting them converged, `status` shows until when they are held, and a tended node waits for the next opening (a `force` still goes through). Teardowns are not held. DAY is Mon..Sun; a start later than the end crosses midnight; the zone is a fixed offset, UTC by default."             <*> overrideWindowP+            <*> switch+                ( long "once"+                    <> Options.Applicative.help+                        "End after the first convergence pass (a declaration's, an explicit `converge`'s or a fetched document's) and exit with its verdict: 0 when it converged (nothing failed, nothing left), 1 when a node failed, was blocked or remains, or when the session ended before any pass ran. Input queued behind the command that ran the pass is not read. Starts with supervision off."+                )+            <*> optional+                ( Options.Applicative.option+                    Options.Applicative.auto+                    ( long "linger"+                        <> Options.Applicative.metavar "SECONDS"+                        <> Options.Applicative.help+                            "With --once: after the pass, keep --listen, --http and --http-tcp up this long (reads answer from the final state, commands are refused) before exiting, so a client such as salmon-tui can show the end state."+                    )+                )+    startupOptionsP =+        StartupOptions+            <$> ( not+                    <$> switch+                        ( long "no-supervise"+                            <> Options.Applicative.help+                                "Start with `supervise off`: nodes are left alone between passes until a `supervise on`."+                        )+                )+            <*> many+                ( strOption+                    ( long "up-directive"+                        <> Options.Applicative.metavar "FILE"+                        <> Options.Applicative.help+                            "Declare the directive JSON in FILE before reading any input, as a first `up-directive FILE` would (repeatable, in order; converged at once unless --no-autoconverge). The file must exist at start-up. With --once the first directive's pass is the one pass: later directives and stdin are not read."+                    )+                )     tcpOptionsP =         TcpOptions             <$> optional@@ -672,6 +755,13 @@             ( long "override-window"                 <> Options.Applicative.help "Run disruptive nodes whatever the maintenance windows say."             )+    jsonLogP =+        ReportTextJsonLog+            <$> strOption+                ( long "json-log"+                    <> Options.Applicative.metavar "FILE"+                    <> Options.Applicative.help "Keep the text report on the terminal and also append every report, as the one-JSON-object-per-line --json prints, to FILE (created if missing, flushed per report). Not with --json."+                )     reportFormatP =         Options.Applicative.flag             ReportText@@ -829,10 +919,34 @@             void $ withGraph (\op -> computedTreeDag op >>= Help.printDagTree)         (Run RunDAG) -> do             void $ withGraph (\op -> computedTreeDag (injectRemoteSubgraphs 0 op) >>= Dot.printDagCograph)-        (Run (RunServe maxConcurrency noAutoConverge fmt followDir labels followOptions listen http eventsRing sinkOptions tcpOptions httpReadOnly wins overrideWins)) -> do+        (Run (RunServe maxConcurrency startup noAutoConverge fmt followDir labels followOptions listen http eventsRing sinkOptions tcpOptions httpReadOnly wins overrideWins once linger)) -> do             reportByLine+            -- a directive named on the command line is looked for before+            -- anything is bound, rather than reported as a bad seed by a+            -- loop that then waits for input nobody is going to type+            forM_ startup.startupDirectives $ \path -> do+                there <- doesFileExist path+                when (not there) $ do+                    hPutStrLn stderr ("--up-directive " <> path <> ": no such file")+                    exitFailure+            -- like --http-read-only without a listener: a flag that would do+            -- nothing is refused rather than ignored+            forM_ linger $ \secs -> do+                when (not once) $ do+                    hPutStrLn stderr "--linger needs --once: it is how long a one-pass session stays up after its pass"+                    exitFailure+                when (secs < 0) $ do+                    hPutStrLn stderr "--linger takes a number of seconds, 0 or more"+                    exitFailure+            -- the first pass's 'Serve.ConvergeStop', which is what --once+            -- exits on; later ones (there are none under --once) are ignored+            verdict <- newIORef Nothing+            let verdictR = ReporterM $ \rep -> case Serve.attributed rep of+                    Serve.ConvergeStop ok remaining -> modifyIORef' verdict (<|> Just (ok, remaining))+                    _ -> pure ()             limit <- traverse Concurrency.newConcurrencyLimit maxConcurrency-            let own = taggedFor fmt+            jsonLog <- traverse openJsonLog (jsonLogPath fmt)+            let own = maybe id (\h t -> reportBoth t (Tagged.reportJSONLines h)) jsonLog (taggedFor fmt)             -- a build without the HTTP server says so before anything is             -- bound or read, rather than after the socket listener is up             when (not httpServerAvailable && (isJust http || isJust tcpOptions.tcpBind)) $ do@@ -924,6 +1038,9 @@             -- registry says, deterministically; after that both interleave             -- at line granularity.             gate <- newEmptyMVar+            -- and before that, the start-up lines: every producer waits+            -- for them to be in the inbox (there are none by default)+            startupGate <- newEmptyMVar             -- what `fetch` pokes: the fetcher's clock wakes on it; and             -- what `status` reads: the fetcher's mode             pk <- Scheduler.newPoke@@ -953,9 +1070,12 @@                       (Nothing, []) -> Serve.stdinProducer stdin                       _ -> Serve.handleProducer (Serve.Origin "stdin") stdin                   producersWith followR more =-                      case follow of+                      startupFirst $ case follow of                           Nothing -> stdinP : more                           Just f -> Follow.follower followR pk modeVar appliedVar f (putMVar gate ()) : Follow.gated gate stdinP : more+                  startupFirst ps = case startupLines startup of+                      [] -> ps+                      ls -> Serve.linesProducer (Serve.Origin "startup") ls (putMVar startupGate ()) : map (Follow.gated startupGate) ps               -- the listener's reporters answer each socket client on its own               -- connection, the HTTP server's answer each request with its               -- own reports, and both hand everything on to the loop's own,@@ -982,14 +1102,15 @@                         nodeLogR = maybe id (\srv nr -> reportBoth nr (Http.serverNodeLogReporter srv)) mserver (Tagged.nodeLogStream tagged)                         saidR = maybe id (\listener sr -> reportBoth sr (Socket.listenerNodeLogReporter listener)) mlistener (contramap Serve.attributed nodeLogR)                     void $-                        Serve.serveSayingWith+                        Serve.serveSayingEnding+                            (if once then Serve.AfterFirstPass else Serve.UntilQuit)                             (if overrideWins then Window.noWindows else Window.windowsAt (rights (map Window.parseWindow wins)))                             saidR                             observe                             rewrites                             limit                             (not noAutoConverge)-                            serveR'+                            (if once then reportBoth serveR' verdictR else serveR')                             r'                             parseSeedArgs                             genBase@@ -997,6 +1118,22 @@                             onFetch                             -- the fetcher's reports also go to /events, as its own stream                             (producersWith (maybe id (\srv r -> reportBoth r (Http.serverFollowReporter srv)) mserver (Tagged.followStream tagged)) more)+                    -- the loop is over and its producers are gone, so the+                    -- listeners answer reads from the final world and+                    -- refuse commands; they close when these brackets end+                    forM_ linger $ \secs -> when (once && secs > 0) $ do+                        hPutStrLn stderr ("serve: lingering " <> show secs <> "s before exiting")+                        threadDelay (secs * 1000000)+            when once $ do+                outcome <- readIORef verdict+                case outcome of+                    Just (True, 0) -> hPutStrLn stderr "serve: --once: converged"+                    Just (ok, remaining) -> do+                        hPutStrLn stderr ("serve: --once: did not converge (" <> show remaining <> " node(s) left" <> (if ok then ")" else ", including a failure)"))+                        exitFailure+                    Nothing -> do+                        hPutStrLn stderr "serve: --once: the session ended before any convergence pass ran"+                        exitFailure         (Query (QueryShow (QuerySelection sel exc) dedupe showDescriptions)) -> do             void $ withGraph $ \op -> do                 let cograph = runIdentity (expand op)@@ -1043,6 +1180,26 @@     taggedFor fmt = case fmt of         ReportText -> Tagged.reportTexts serveR r silent Follow.reportText (Tagged.reportNodeLogText stdout)         ReportJson -> Tagged.reportJSONLines stdout+        -- the JSON half needs a file opened, which @serve@ adds beside this+        ReportTextJsonLog _ -> taggedFor ReportText++    jsonLogPath :: ReportFormat -> Maybe FilePath+    jsonLogPath (ReportTextJsonLog path) = Just path+    jsonLogPath _ = Nothing++    {- | @--json-log FILE@, opened for appending before anything is bound or+    read, so a path that cannot be written stops the loop before it starts.+    The handle stays open for the life of the process: every report is+    flushed as it is written ('Tagged.reportJSONLines'), so nothing is lost+    to its not being closed. -}+    openJsonLog :: FilePath -> IO Handle+    openJsonLog path = do+        opened <- try (openFile path AppendMode)+        case opened of+            Left err -> do+                hPutStrLn stderr ("--json-log " <> path <> ": " <> show (err :: IOException))+                exitFailure+            Right h -> pure h      -- | The tagged reporter split contravariantly into the two the drivers take.     reportersOver :: Reporter Tagged.Tagged -> (Reporter Serve.Report, Reporter (UpDown.Report Extension))

@@ -0,0 +1,649 @@+{- | A ClickHouse server and what lives in it: the server as a podman quadlet,+its configuration drop-ins, and databases, users and grants made with SQL.++Everything that talks to a server goes through @clickhouse-client@, either+the host's ('LocalClient') or the container's own through @podman exec@+('InContainer'), as an administrator that needs no password: the+@default@ user, which 'renderAdminAccess' gives SQL access management and+confines to the loopback addresses. A server administered over the network,+or with an administrator password, is not covered.++Nothing here reads a secret at graph build. A user's password is a file+('userPassFile') read at @up@ and @check@ time; what @up@ sends is a salted+SHA-256 of it, on @clickhouse-client@'s standard input (ClickHouse masks it in+@system.query_log@), and what @check@ does is log in as the user with the+password in @CLICKHOUSE_PASSWORD@, in the client's environment, never its+argv. Neither the password nor the hash is in a report, a failure or a node's+@help@ or @notes@.++Nothing here drops anything: @down@ leaves databases, users and grants in+place (the server's own @down@ stops the container and leaves its data+directory).+-}+module Salmon.Builtin.Nodes.ClickHouse where++import Control.Exception (Exception, IOException, throwIO, try)+import qualified Crypto.Hash as Hash+import qualified Crypto.Random as Random+import qualified Data.ByteString as ByteString+import Data.Char (isAlphaNum, isAscii, isDigit, toUpper)+import Data.List (sort)+import Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.Encoding as Text+import qualified Data.Text.Encoding.Error as TextError+import qualified Data.Text.IO as Text+import System.Environment (getEnvironment)+import System.Exit (ExitCode (..))+import System.FilePath ((</>))+import System.Process.ByteString (readCreateProcessWithExitCode)+import System.Process.ListLike (CreateProcess (..), proc)++import Salmon.Actions.UpDown (CheckResult (..))+import Salmon.Builtin.Extension+import Salmon.Builtin.Nodes.Binary (Binary, Command (..), justInstall, withBinaryStdin)+import qualified Salmon.Builtin.Nodes.Binary as Binary+import Salmon.Builtin.Nodes.Filesystem (File, withFile)+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 qualified Salmon.Builtin.Nodes.Systemd as Systemd+import Salmon.Op.Ref+import Salmon.Op.Track+import Salmon.Reporter++-------------------------------------------------------------------------------++data Report+    = CHCreateDatabase !DatabaseName !Binary.Report+    | CHSetUser !UserName !Binary.Report+    | CHGrant !Grant !Binary.Report+    | CHSourceGrant !Text !UserName !Binary.Report+    deriving (Show)++type DatabaseName = Text+type UserName = Text++-------------------------------------------------------------------------------+-- Configuration drop-ins++{- | The server settings salmon renders into one @config.d@ drop-in. ClickHouse+merges every file of @config.d@ over its own @config.xml@, so a setting left+out here is the packaged default.+-}+data ServerConfig+    = ServerConfig+    { serverListenHosts :: [Text]+    -- ^ @listen_host@, one element each; empty renders none (the packaged+    -- default, which in the official image is every address)+    , serverHttpPort :: Maybe Int+    , serverTcpPort :: Maybe Int+    , serverSettings :: [(Text, Text)]+    -- ^ other top-level scalar settings, as @(element, value)@+    -- (@("max_connections", "2048")@); the value is escaped, the element must+    -- be a plain XML name ('configProblems')+    }+    deriving (Eq, Ord, Show)++-- | Renders nothing but the root element: every setting is the packaged default.+defaultServerConfig :: ServerConfig+defaultServerConfig = ServerConfig [] Nothing Nothing []++-- | Native protocol port, the one @clickhouse-client@ speaks.+defaultTcpPort :: Int+defaultTcpPort = 9000++-- | HTTP interface port.+defaultHttpPort :: Int+defaultHttpPort = 8123++renderServerConfig :: ServerConfig -> Text+renderServerConfig cfg =+    Text.unlines $+        mconcat+            [ ["<clickhouse>"]+            , [element "listen_host" h | h <- cfg.serverListenHosts]+            , [element "http_port" (showT p) | Just p <- [cfg.serverHttpPort]]+            , [element "tcp_port" (showT p) | Just p <- [cfg.serverTcpPort]]+            , [element k v | (k, v) <- cfg.serverSettings]+            , ["</clickhouse>"]+            ]+  where+    element k v = "    <" <> k <> ">" <> xmlEscape v <> "</" <> k <> ">"++{- | What makes a 'ServerConfig' unrenderable: a setting whose element is not a+plain XML name, which would otherwise be written as markup.+-}+configProblems :: ServerConfig -> [Text]+configProblems cfg =+    ["setting " <> k <> " is not a plain XML element name" | (k, _) <- cfg.serverSettings, not (plainName k)]+  where+    plainName k = case Text.uncons k of+        Just (c, rest) -> (isAscii c && (c == '_' || isAlphaNum c) && not (isDigit c)) && Text.all (\x -> isAscii x && (x == '_' || isAlphaNum x)) rest+        Nothing -> False++{- | The @users.d@ drop-in that makes the @default@ user the administrator+the SQL nodes use: @access_management@ (so it may @CREATE USER@ and @GRANT@),+and reachable from the loopback addresses only. The packaged @users.xml@+lets @default@ in from anywhere with an empty password, so a server whose+port is published without this file is open to whoever reaches it.+-}+renderAdminAccess :: Text+renderAdminAccess =+    Text.unlines+        [ "<clickhouse>"+        , "    <users>"+        , "        <default>"+        , "            <access_management>1</access_management>"+        , "            <networks replace=\"replace\">"+        , "                <ip>::1</ip>"+        , "                <ip>127.0.0.1</ip>"+        , "            </networks>"+        , "        </default>"+        , "    </users>"+        , "</clickhouse>"+        ]++xmlEscape :: Text -> Text+xmlEscape = Text.concatMap esc+  where+    esc '&' = "&amp;"+    esc '<' = "&lt;"+    esc '>' = "&gt;"+    esc '"' = "&quot;"+    esc '\'' = "&apos;"+    esc c = Text.singleton c++data InvalidConfig = InvalidConfig !FilePath ![Text]+    deriving (Show)++instance Exception InvalidConfig++{- | One drop-in file, written in a directory shared with files salmon does+not declare (@\/etc\/clickhouse-server\/config.d@ on a packaged install).+A 'ServerConfig' with 'configProblems' fails at @up@ and writes nothing.+-}+serverConfigFile :: FilePath -> ServerConfig -> Op+serverConfigFile path cfg =+    case configProblems cfg of+        [] -> FS.filecontentsInSharedDir (FS.FileContents path (renderServerConfig cfg))+        problems ->+            op "clickhouse-config" nodeps $ \actions ->+                actions+                    { help = "refuses to write " <> Text.pack path+                    , notes = problems+                    , ref = mkRef "file-contents" path+                    , up = throwIO (InvalidConfig path problems)+                    }++-- | 'renderAdminAccess' at this path (a @users.d@ drop-in).+adminAccessFile :: FilePath -> Op+adminAccessFile path = FS.filecontentsInSharedDir (FS.FileContents path renderAdminAccess)++-------------------------------------------------------------------------------+-- The server as a quadlet++-- | A ClickHouse server run from the official image as a system-scope quadlet.+data Server+    = Server+    { serverName :: Podman.ContainerName+    , serverImage :: Text+    -- ^ the full reference, e.g. @docker.io\/clickhouse\/clickhouse-server:25.8@+    , serverConfigDir :: FilePath+    -- ^ a host directory salmon owns, holding the two drop-ins, which are+    -- bind-mounted read-only into the container's @config.d@ and @users.d@+    , serverDataDir :: FilePath+    -- ^ the host directory mounted at @\/var\/lib\/clickhouse@, created if+    -- missing; the image's entrypoint gives it to the @clickhouse@ user at+    -- start. Never removed.+    , serverConfig :: ServerConfig+    , serverPorts :: [Podman.PortMapping]+    -- ^ published ports; empty publishes none (the SQL nodes reach the server+    -- through @podman exec@ either way)+    , serverReadyTimeout :: Int+    -- ^ seconds @up@ waits for the server to answer a query after a start+    }+    deriving (Eq, Show)++-- | No port published, the default configuration, 60 seconds to answer.+server :: Podman.ContainerName -> Text -> FilePath -> FilePath -> Server+server name image configDir dataDir =+    Server+        { serverName = name+        , serverImage = image+        , serverConfigDir = configDir+        , serverDataDir = dataDir+        , serverConfig = defaultServerConfig+        , serverPorts = []+        , serverReadyTimeout = 60+        }++serverConfigPath :: Server -> FilePath+serverConfigPath s = s.serverConfigDir </> "salmon-server.xml"++adminAccessPath :: Server -> FilePath+adminAccessPath s = s.serverConfigDir </> "salmon-admin.xml"++-- | The client the SQL nodes use for this server: its own, in the container.+serverClient :: Server -> Client+serverClient s = Client (InContainer s.serverName) (maybe defaultTcpPort id s.serverConfig.serverTcpPort)++{- | The quadlet: the image, the data directory, the two drop-ins (watched, so+a changed one restarts the container), and a readiness probe asking the+server for @SELECT 1@ through its own client.+-}+serverContainer :: Server -> Quadlet.Container+serverContainer s =+    (Quadlet.container s.serverName s.serverImage)+        { Quadlet.containerDescription = "ClickHouse " <> Podman.getContainerName s.serverName <> " (salmon)"+        , Quadlet.containerPorts = s.serverPorts+        , Quadlet.containerBinds =+            [ Quadlet.bind s.serverDataDir "/var/lib/clickhouse"+            , readOnlyFile (serverConfigPath s) "/etc/clickhouse-server/config.d/salmon-server.xml"+            , readOnlyFile (adminAccessPath s) "/etc/clickhouse-server/users.d/salmon-admin.xml"+            ]+        , Quadlet.containerWatched = [serverConfigPath s, adminAccessPath s]+        , Quadlet.containerReady = Just (Quadlet.readyWhen (readyProbe (serverClient s)) s.serverReadyTimeout)+        }+  where+    readOnlyFile host guest =+        (Quadlet.bind host guest){Quadlet.bindMode = Podman.ReadOnly, Quadlet.bindCreate = Nothing}++-- | @SELECT 1@ through this client exits 0.+readyProbe :: Client -> Readiness.Probe+readyProbe c = case clientArgv c ["--query", "SELECT 1"] of+    (exe : args) -> Readiness.ProbeCommand exe (fmap Text.pack args)+    [] -> Readiness.ProbeCommand "false" []++{- | The server: its drop-ins written, then the quadlet ('Quadlet.quadletContainer'),+whose @up@ returns once the server answers. Use it as the @Track'@ of the+SQL nodes' client: @Track (const (serverOp r systemctl s))@.+-}+serverOp :: Reporter Systemd.Report -> Track' (Binary "systemctl") -> Server -> Op+serverOp r systemctl s =+    Quadlet.quadletContainer r systemctl (Track (const files)) (serverContainer s)+  where+    files =+        op "clickhouse-config-files" (deps [serverConfigFile (serverConfigPath s) s.serverConfig, adminAccessFile (adminAccessPath s)]) $ \actions ->+            actions{help = "ClickHouse drop-ins in " <> Text.pack s.serverConfigDir}++-------------------------------------------------------------------------------+-- The client++-- | Where @clickhouse-client@ runs.+data Via+    = -- | the host's, against the loopback address+      LocalClient+    | -- | the container's own, through @podman exec@+      InContainer Podman.ContainerName+    deriving (Eq, Ord, Show)++data Client = Client {clientVia :: Via, clientPort :: Int}+    deriving (Eq, Ord, Show)++-- | The host's client against @localhost:9000@.+localClient :: Client+localClient = Client LocalClient defaultTcpPort++-- | Which server a node is about, for its 'Ref' and its @notes@.+clientKey :: Client -> Text+clientKey c = case c.clientVia of+    LocalClient -> "localhost:" <> showT c.clientPort+    InContainer name -> "container:" <> Podman.getContainerName name <> ":" <> showT c.clientPort++{- | The argv of a client call: @clickhouse-client --port P@ and these+arguments, behind @podman exec -i NAME@ for a container. Statements go on+standard input, not here.+-}+clientArgv :: Client -> [String] -> [String]+clientArgv c args = case c.clientVia of+    LocalClient -> base+    InContainer name -> ["podman", "exec", "-i", Text.unpack (Podman.getContainerName name)] <> base+  where+    base = ["clickhouse-client", "--port", show c.clientPort] <> args++-- | The administrator's batch: every statement read from standard input.+data AdminBatch = AdminBatch+    deriving (Show)++adminRun :: Client -> Command "clickhouse-client" AdminBatch+adminRun c = Command $ \AdminBatch -> argvProc (clientArgv c ["--multiquery"])++argvProc :: [String] -> CreateProcess+argvProc (exe : args) = proc exe args+argvProc [] = proc "false" []++{- | One query as the administrator, its output back; 'Left' with the+client's error when it fails. For the checks: the query names databases and+users, never a secret.+-}+adminQuery :: Client -> Text -> IO (Either Text Text)+adminQuery c sql = do+    r <- try (readCreateProcessWithExitCode (argvProc (clientArgv c [])) (Text.encodeUtf8 sql)) :: IO (Either IOException (ExitCode, ByteString.ByteString, ByteString.ByteString))+    pure $ case r of+        Left e -> Left (Text.pack (show e))+        Right (ExitSuccess, out, _) -> Right (decode out)+        Right (ExitFailure _, _, err) -> Left (Text.strip (decode err))++-------------------------------------------------------------------------------+-- SQL quoting++-- | A backquoted identifier; a backquote or backslash in the name is escaped.+quoteIdent :: Text -> Text+quoteIdent t = "`" <> Text.concatMap esc t <> "`"+  where+    esc '`' = "\\`"+    esc '\\' = "\\\\"+    esc c = Text.singleton c++-- | A single-quoted string literal; a quote or backslash in it is escaped.+quoteLiteral :: Text -> Text+quoteLiteral t = "'" <> Text.concatMap esc t <> "'"+  where+    esc '\'' = "\\'"+    esc '\\' = "\\\\"+    esc c = Text.singleton c++-------------------------------------------------------------------------------+-- Databases++{- | A database, created if missing (engine left to the server, @Atomic@).+@down@ drops nothing.+-}+database :: Reporter Report -> Track' (Binary "clickhouse-client") -> Client -> DatabaseName -> Op+database r clientT c name =+    withBinaryStdin clientT (adminRun c) AdminBatch (Text.encodeUtf8 (createDatabaseSql name)) $ \create ->+        op "clickhouse-database" nodeps $ \actions ->+            actions+                { ref = mkRef "clickhouse-db" (clientKey c, name)+                , help = "create ClickHouse database " <> name+                , notes = ["server: " <> clientKey c]+                , check = checkDatabase c name+                , up = create (contramap (CHCreateDatabase name) r)+                }++createDatabaseSql :: DatabaseName -> Text+createDatabaseSql name = "CREATE DATABASE IF NOT EXISTS " <> quoteIdent name <> ";\n"++databaseExistsSql :: DatabaseName -> Text+databaseExistsSql name = "SELECT count() FROM system.databases WHERE name = " <> quoteLiteral name++checkDatabase :: Client -> DatabaseName -> IO CheckResult+checkDatabase c name = either (const Unknown) (interpretCount ("database " <> name <> " does not exist")) <$> adminQuery c (databaseExistsSql name)++-- | @1@ is 'Success', @0@ a 'Failure' with this text, anything else 'Unknown'.+interpretCount :: Text -> Text -> CheckResult+interpretCount missing out = case Text.strip out of+    "1" -> Success+    "0" -> Failure missing+    _ -> Unknown++-------------------------------------------------------------------------------+-- Users++newtype Password = Password {revealPassword :: Text}++instance Show Password where+    show _ = "<password>"++{- | A user whose password is the content of a file, read at @check@ and @up@+time (never at graph build), verbatim (a trailing newline is part of it, as+for "Salmon.Builtin.Nodes.Postgres".'Salmon.Builtin.Nodes.Postgres.userPassFile').++@up@ creates the user if missing and sets the password either way, so a+rotated file reaches the server: @CREATE USER IF NOT EXISTS@ then @ALTER USER+... IDENTIFIED WITH sha256_hash BY HASH SALT SALT@, the hash of the password+and a fresh random salt, on the client's standard input. The user may log in+from any host. Its grants are left alone.++The check is 'checkUser': the user exists, and a login as it with the file's+password succeeds.+-}+userPassFile :: Reporter Report -> Track' (Binary "clickhouse-client") -> Client -> File "passfile" -> UserName -> Op+userPassFile r clientT c passfile name =+    withFile passfile $ \path ->+        op "clickhouse-user" (deps [justInstall clientT]) $ \actions ->+            actions+                { ref = mkRef "clickhouse-user" (clientKey c, name)+                , help = Text.unwords ["set ClickHouse user", name, "password from file at", Text.pack path]+                , notes = ["server: " <> clientKey c]+                , check = checkUser c name (readPassword path)+                , up = do+                    pwd <- readPassword path+                    salt <- Random.getRandomBytes saltBytes+                    setUser (adminRun c) name pwd salt (contramap (CHSetUser name) r)+                }++readPassword :: FilePath -> IO Password+readPassword path = Password <$> Text.readFile path++saltBytes :: Int+saltBytes = 16++-- | ClickHouse's @sha256_hash@: the hex SHA-256 of the password followed by the salt.+sha256Hash :: Password -> Text -> Text+sha256Hash pwd salt =+    Text.pack (show (Hash.hashWith Hash.SHA256 (Text.encodeUtf8 (pwd.revealPassword <> salt))))++hexSalt :: ByteString.ByteString -> Text+hexSalt = Text.pack . concatMap byte . ByteString.unpack+  where+    byte w = [digits !! fromIntegral (w `div` 16), digits !! fromIntegral (w `mod` 16)]+    digits = "0123456789abcdef"++{- | The statements 'userPassFile' sends, for this password and salt. Holds a+digest of the password: never reported, only written to the client's stdin.+-}+userSql :: UserName -> Password -> ByteString.ByteString -> Text+userSql name pwd saltRaw =+    Text.unlines+        [ "CREATE USER IF NOT EXISTS " <> quoteIdent name <> identified <> ";"+        , "ALTER USER " <> quoteIdent name <> identified <> ";"+        ]+  where+    salt = hexSalt saltRaw+    identified = " IDENTIFIED WITH sha256_hash BY " <> quoteLiteral (sha256Hash pwd salt) <> " SALT " <> quoteLiteral salt++-- | Thrown by a failed 'setUser', with the hash and salt cut out of the client's output.+data SetUserFailed = SetUserFailed !UserName !Int !Text+    deriving (Show)++instance Exception SetUserFailed++{- | Runs 'userSql'. Not 'Binary.untrackedExec': its failure carries the+client's output, and a ClickHouse syntax error quotes the statement, which+holds the hash. Here the reported and thrown output has the hash and the salt+replaced by @<redacted>@.+-}+setUser :: Command "clickhouse-client" AdminBatch -> UserName -> Password -> ByteString.ByteString -> Reporter Binary.Report -> IO ()+setUser cmd name pwd saltRaw r = do+    let p = prepare cmd AdminBatch+        salt = hexSalt saltRaw+        redact = redactAll [sha256Hash pwd salt, salt]+    runReporter r (Binary.CommandStart p)+    (code, out, err) <- readCreateProcessWithExitCode p (Text.encodeUtf8 (userSql name pwd saltRaw))+    let out' = redactBytes redact out+        err' = redactBytes redact err+    runReporter r (Binary.CommandStopped p code out' err')+    case code of+        ExitSuccess -> pure ()+        ExitFailure n -> throwIO (SetUserFailed name n (decode err'))++redactAll :: [Text] -> Text -> Text+redactAll secrets t = foldr (\s acc -> if Text.null s then acc else Text.replace s "<redacted>" acc) t secrets++redactBytes :: (Text -> Text) -> ByteString.ByteString -> ByteString.ByteString+redactBytes f = Text.encodeUtf8 . f . decode++userExistsSql :: UserName -> Text+userExistsSql name = "SELECT count() FROM system.users WHERE name = " <> quoteLiteral name++{- | The check of a @clickhouse-user@ node. ClickHouse never gives a password+or its hash back (@system.users@ shows the scheme only), so the password is+tried: a login as the user, with the password in @CLICKHOUSE_PASSWORD@ in the+client's environment (through @podman exec -e@ for a container), running+@SELECT currentUser()@.++* the file cannot be read, or the server cannot be asked: 'Unknown'+* no such user: 'Failure'+* the login is refused (@AUTHENTICATION_FAILED@): 'Failure'+* the login works: 'Success'++Nothing secret is in a verdict. A refused login is in the server's log, as+every wrong password is.+-}+checkUser :: Client -> UserName -> IO Password -> IO CheckResult+checkUser c name readPass = do+    pass <- try readPass :: IO (Either IOException Password)+    case pass of+        Left _ -> pure Unknown+        Right pwd -> do+            exists <- adminQuery c (userExistsSql name)+            case interpretCount ("user " <> name <> " does not exist") <$> exists of+                Left _ -> pure Unknown+                Right Success -> loginAs c name pwd+                Right other -> pure other++loginAs :: Client -> UserName -> Password -> IO CheckResult+loginAs c name pwd = do+    environment <- getEnvironment+    let password = Text.unpack pwd.revealPassword+        argv = loginArgv c name+        p = (argvProc argv){env = Just (("CLICKHOUSE_PASSWORD", password) : filter ((/= "CLICKHOUSE_PASSWORD") . fst) environment)}+    r <- try (readCreateProcessWithExitCode p "") :: IO (Either IOException (ExitCode, ByteString.ByteString, ByteString.ByteString))+    pure $ case r of+        Left _ -> Unknown+        Right (code, _, err) -> interpretLogin name code (decode err)++{- | @clickhouse-client --user NAME --query 'SELECT currentUser()'@; for a+container, @podman exec -e CLICKHOUSE_PASSWORD@, which passes podman's own+variable in, so the value is on no command line.+-}+loginArgv :: Client -> UserName -> [String]+loginArgv c name = case c.clientVia of+    LocalClient -> base+    InContainer cname -> ["podman", "exec", "-i", "-e", "CLICKHOUSE_PASSWORD", Text.unpack (Podman.getContainerName cname)] <> base+  where+    base = ["clickhouse-client", "--port", show c.clientPort, "--user", Text.unpack name, "--query", "SELECT currentUser()"]++interpretLogin :: UserName -> ExitCode -> Text -> CheckResult+interpretLogin _ ExitSuccess _ = Success+interpretLogin name (ExitFailure _) err+    | "AUTHENTICATION_FAILED" `Text.isInfixOf` err = Failure ("ClickHouse refused the declared password of user " <> name)+    | otherwise = Unknown++-------------------------------------------------------------------------------+-- Grants++{- | Privileges on every table of one database, for one user (or role).++Name the privileges as ClickHouse lists them in @SHOW GRANTS@ (@SELECT@,+@INSERT@, @ALTER@, @CREATE TABLE@, @DROP TABLE@, ...). A compound one such as+@ALL@ or @CREATE@ is stored expanded into what it holds, so the check never+finds it as written and @up@ runs again each pass (harmless, a @GRANT@ is+idempotent, but never satisfied).+-}+data Grant+    = Grant+    { grantPrivileges :: [Text]+    , grantDatabase :: DatabaseName+    , grantTo :: UserName+    }+    deriving (Eq, Ord, Show)++{- | @GRANT privs ON db.* TO user@. The check is 'checkGrant'. @down@ revokes+nothing. The user and the database must exist first: declare them as+predecessors (@grant ... \`inject\` userOp@), the node does not.+-}+grant :: Reporter Report -> Track' (Binary "clickhouse-client") -> Client -> Grant -> Op+grant r clientT c g =+    withBinaryStdin clientT (adminRun c) AdminBatch (Text.encodeUtf8 (grantSql g)) $ \apply ->+        op "clickhouse-grant" nodeps $ \actions ->+            actions+                { ref = mkRef "clickhouse-grant" (clientKey c, g.grantTo, g.grantDatabase, sort (normalizedPrivileges g))+                , help = Text.unwords ["grant", Text.intercalate ", " (normalizedPrivileges g), "on", g.grantDatabase <> ".*", "to", g.grantTo]+                , notes = ["server: " <> clientKey c]+                , check = checkGrant c g+                , up = apply (contramap (CHGrant g) r)+                }++normalizedPrivileges :: Grant -> [Text]+normalizedPrivileges g = fmap (Text.unwords . Text.words . Text.map toUpper) g.grantPrivileges++grantSql :: Grant -> Text+grantSql g =+    "GRANT "+        <> Text.intercalate ", " (normalizedPrivileges g)+        <> " ON "+        <> quoteIdent g.grantDatabase+        <> ".* TO "+        <> quoteIdent g.grantTo+        <> ";\n"++-- | The database-wide privileges the grantee holds on the database, one per line.+grantedSql :: Grant -> Text+grantedSql g =+    "SELECT access_type FROM system.grants WHERE (user_name = "+        <> quoteLiteral g.grantTo+        <> " OR role_name = "+        <> quoteLiteral g.grantTo+        <> ") AND database = "+        <> quoteLiteral g.grantDatabase+        <> " AND table IS NULL AND column IS NULL AND is_partial_revoke = 0"++checkGrant :: Client -> Grant -> IO CheckResult+checkGrant c g = either (const Unknown) (interpretGrants g) <$> adminQuery c (grantedSql g)++-- | 'Success' when every declared privilege is among the lines, compared without case.+interpretGrants :: Grant -> Text -> CheckResult+interpretGrants g out =+    case filter (`notElem` held) (normalizedPrivileges g) of+        [] -> Success+        missing -> Failure (g.grantTo <> " lacks " <> Text.intercalate ", " missing <> " on " <> g.grantDatabase <> ".*")+  where+    held = fmap (Text.map toUpper . Text.strip) (filter (not . Text.null . Text.strip) (Text.lines out))++{- | @GRANT READ ON SOURCE TO user@: leave to read an external source, such as+@S3@ (what a PeerDB ClickHouse peer needs: it stages rows in S3 and has+ClickHouse read them, and PeerDB refuses the peer without it). The source+name is ClickHouse's (@S3@, @URL@, @FILE@, ...), upper case. Checked on the+@system.grants@ row (@access_type = READ@, @access_object@ the source), as+25.8 stores it. @down@ revokes nothing. Inject the user first.+-}+sourceReadGrant :: Reporter Report -> Track' (Binary "clickhouse-client") -> Client -> Text -> UserName -> Op+sourceReadGrant r clientT c rawSource name =+    withBinaryStdin clientT (adminRun c) AdminBatch (Text.encodeUtf8 (sourceReadGrantSql source name)) $ \apply ->+        op "clickhouse-source-grant" nodeps $ \actions ->+            actions+                { ref = mkRef "clickhouse-source-grant" (clientKey c, name, source)+                , help = Text.unwords ["grant READ on", source, "to", name]+                , notes = ["server: " <> clientKey c]+                , check = either (const Unknown) (interpretCount (name <> " may not read " <> source)) <$> adminQuery c (sourceReadGrantedSql source name)+                , up = apply (contramap (CHSourceGrant source name) r)+                }+  where+    source = sourceName rawSource++-- | A source as ClickHouse names it: upper case, letters and digits only.+sourceName :: Text -> Text+sourceName = Text.filter isAlphaNum . Text.toUpper++sourceReadGrantSql :: Text -> UserName -> Text+sourceReadGrantSql source name = "GRANT READ ON " <> sourceName source <> " TO " <> quoteIdent name <> ";\n"++sourceReadGrantedSql :: Text -> UserName -> Text+sourceReadGrantedSql source name =+    "SELECT count() > 0 FROM system.grants WHERE user_name = "+        <> quoteLiteral name+        <> " AND access_type = 'READ' AND access_object = "+        <> quoteLiteral (sourceName source)+        <> " AND is_partial_revoke = 0"++-------------------------------------------------------------------------------++showT :: (Show a) => a -> Text+showT = Text.pack . show++decode :: ByteString.ByteString -> Text+decode = Text.decodeUtf8With TextError.lenientDecode

@@ -0,0 +1,553 @@+{- | PeerDB peers and CDC mirrors, declared through PeerDB's flow API (the+HTTP/JSON gateway of @flow-api@, port 8113 upstream).++What this module covers is what PeerDB holds: a Postgres source peer, a+ClickHouse destination peer, and a CDC mirror between them. PeerDB itself+(its catalog, Temporal, @flow-api@ and the workers, and the S3-compatible+store ClickHouse mirrors stage through) is not run here: the caller gives a+'Track'' 'Api' that stands for it being up.++The flow API is spoken in plain HTTP with no credential: run it where only+the operator reaches it. A @flow-api@ that wants a password, or TLS, is not+covered.++Secrets. A peer's password is a file, read at @up@ only (the check cannot+use it: PeerDB gives every password back as @********@) and sent in the+request body, never in a URL, a report, a failure or a node's @help@ or+@notes@. What PeerDB answers is cut of the password before it is reported or+thrown.++Nothing here drops anything: @down@ of a peer or a mirror is nothing.+Dropping a mirror is PeerDB's @TERMINATED@ state change, which removes its+slot and publication on the source and, unless told not to, its tables on+the destination: an operator's decision, as is a resync.+-}+module Salmon.Builtin.Nodes.PeerDB where++import Control.Exception (Exception, throwIO, try)+import Data.Aeson (Value (..), object, (.=))+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Key as Key+import qualified Data.Aeson.KeyMap as KeyMap+import qualified Data.ByteString as ByteString+import qualified Data.ByteString.Lazy as LByteString+import Data.Foldable (toList)+import Data.List (sort)+import Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.Encoding as Text+import qualified Data.Text.Encoding.Error as TextError+import qualified Data.Text.IO as Text+import Network.HTTP.Client (+    HttpException,+    Manager,+    RequestBody (..),+    defaultManagerSettings,+    httpLbs,+    managerResponseTimeout,+    method,+    newManager,+    parseRequest,+    requestBody,+    requestHeaders,+    responseBody,+    responseStatus,+    responseTimeoutMicro,+ )+import qualified Network.HTTP.Client as Http+import Network.HTTP.Types.Status (statusCode)+import Network.HTTP.Types.URI (urlEncode)++import Salmon.Actions.UpDown (CheckResult (..))+import Salmon.Builtin.Extension+import Salmon.Builtin.Nodes.Filesystem (File, withFile)+import Salmon.Op.Ref+import Salmon.Op.Track+import Salmon.Reporter++-------------------------------------------------------------------------------++-- | Where the flow API answers, e.g. @http:\/\/127.0.0.1:8113@ (no trailing slash).+newtype Api = Api {apiBaseUrl :: Text}+    deriving (Eq, Ord, Show)++{- | One request: what it was for, the method and path, and the HTTP status+(@0@ when nothing answered). Never a body: a peer's body holds its password.+-}+data Report+    = ApiCall !Text !Text !Int+    deriving (Show)++-------------------------------------------------------------------------------+-- Peers++type PeerName = Text++data PostgresPeer+    = PostgresPeer+    { pgHost :: Text+    , pgPort :: Int+    , pgUser :: Text+    , pgDatabase :: Text+    , pgRequireTls :: Bool+    }+    deriving (Eq, Ord, Show)++data ClickHousePeer+    = ClickHousePeer+    { chHost :: Text+    , chPort :: Int+    -- ^ the native protocol port (9000, or 9440 with TLS)+    , chUser :: Text+    , chDatabase :: Text+    , chDisableTls :: Bool+    }+    deriving (Eq, Ord, Show)++data PeerConfig+    = PostgresSource PostgresPeer+    | ClickHouseTarget ClickHousePeer+    deriving (Eq, Ord, Show)++data Peer = Peer {peerName :: PeerName, peerConfig :: PeerConfig}+    deriving (Eq, Ord, Show)++{- | A peer, created or updated (@allow_update@), with the password read from+this file at @up@, verbatim (a trailing newline is part of it, as for the+user nodes of "Salmon.Builtin.Nodes.Postgres" and+"Salmon.Builtin.Nodes.ClickHouse", which can read the same file).++PeerDB validates a peer before it keeps it: it connects with it, and for a+ClickHouse peer checks the user may read S3 (see+"Salmon.Builtin.Nodes.ClickHouse".'Salmon.Builtin.Nodes.ClickHouse.sourceReadGrant'),+so @up@ fails while the database it names does not let it in.++The check reads the peer back and compares what PeerDB shows (the type,+host, port, user, database and TLS switch). It cannot compare the password,+which PeerDB shows as @********@: a rotated file is sent by the next @up@,+which a passing check skips, so a rotation needs the node forced. @down@ is+nothing.+-}+peer :: Reporter Report -> Track' Api -> Api -> File "passfile" -> Peer -> Op+peer r apiT api passfile p =+    withFile passfile $ \path ->+        op "peerdb-peer" (deps [run apiT api]) $ \actions ->+            actions+                { ref = mkRef "peerdb-peer" (api.apiBaseUrl, p.peerName)+                , help = Text.unwords ["PeerDB", peerTypeName p.peerConfig, "peer", p.peerName, "with the password from file at", Text.pack path]+                , notes = ["flow api: " <> api.apiBaseUrl, peerSummary p.peerConfig]+                , check = checkPeer api p+                , up = do+                    pwd <- Text.readFile path+                    createPeer r api p (Password pwd)+                }++newtype Password = Password {revealPassword :: Text}++instance Show Password where+    show _ = "<password>"++peerTypeName :: PeerConfig -> Text+peerTypeName (PostgresSource _) = "POSTGRES"+peerTypeName (ClickHouseTarget _) = "CLICKHOUSE"++-- | The non-secret part of a peer, as its notes show it.+peerSummary :: PeerConfig -> Text+peerSummary (PostgresSource c) =+    "postgres " <> c.pgUser <> "@" <> c.pgHost <> ":" <> showT c.pgPort <> "/" <> c.pgDatabase <> (if c.pgRequireTls then " tls required" else "")+peerSummary (ClickHouseTarget c) =+    "clickhouse " <> c.chUser <> "@" <> c.chHost <> ":" <> showT c.chPort <> "/" <> c.chDatabase <> (if c.chDisableTls then " no tls" else " tls")++{- | The fields of a peer's configuration, as the flow API names them in a+request. The password is the one secret field.+-}+peerConfigFields :: PeerConfig -> Password -> (Text, Value)+peerConfigFields (PostgresSource c) pwd =+    ( "postgres_config"+    , object+        [ "host" .= c.pgHost+        , "port" .= c.pgPort+        , "user" .= c.pgUser+        , "password" .= pwd.revealPassword+        , "database" .= c.pgDatabase+        , "require_tls" .= c.pgRequireTls+        ]+    )+peerConfigFields (ClickHouseTarget c) pwd =+    ( "clickhouse_config"+    , object+        [ "host" .= c.chHost+        , "port" .= c.chPort+        , "user" .= c.chUser+        , "password" .= pwd.revealPassword+        , "database" .= c.chDatabase+        , "disable_tls" .= c.chDisableTls+        ]+    )++-- | The body of @POST \/v1\/peers\/create@. Holds the password.+createPeerBody :: Peer -> Password -> Value+createPeerBody p pwd =+    object+        [ "peer"+            .= object+                [ "name" .= p.peerName+                , "type" .= peerTypeName p.peerConfig+                , Key.fromText field .= cfg+                ]+        , "allow_update" .= True+        ]+  where+    (field, cfg) = peerConfigFields p.peerConfig pwd++createPeer :: Reporter Report -> Api -> Peer -> Password -> IO ()+createPeer r api p pwd = do+    (code, body) <- call r ("peer " <> p.peerName) api "POST" "/v1/peers/create" (Just (createPeerBody p pwd))+    let answer = redact [pwd.revealPassword] (decode body)+    case (code, jsonField "status" =<< Aeson.decodeStrict body) of+        (200, Just (String "CREATED")) -> pure ()+        _ -> throwIO (PeerDBRefused ("peer " <> p.peerName <> ": " <> statusText code <> errorMessage answer))++{- | @GET \/v1\/peers\/info\/NAME@: 404 is a 'Failure' (no such peer),+another answer is compared by 'interpretPeerInfo', a flow API that does not+answer is 'Unknown'.+-}+checkPeer :: Api -> Peer -> IO CheckResult+checkPeer api p = do+    res <- request api "GET" ("/v1/peers/info/" <> pathSegment p.peerName) Nothing+    pure $ case res of+        Left _ -> Unknown+        Right (404, _) -> Failure ("PeerDB has no peer " <> p.peerName)+        Right (200, body) -> maybe Unknown (interpretPeerInfo p) (Aeson.decodeStrict body)+        Right _ -> Unknown++{- | What @\/v1\/peers\/info@ answers, read against the declaration. The+answer's fields are in lowerCamelCase; the password is not looked at (it is+@********@).+-}+interpretPeerInfo :: Peer -> Value -> CheckResult+interpretPeerInfo p v =+    case jsonField "peer" v of+        Nothing -> Unknown+        Just held -> case differences held of+            [] -> Success+            ds -> Failure ("PeerDB peer " <> p.peerName <> " differs: " <> Text.intercalate ", " ds)+  where+    differences held =+        [ "type is " <> renderValue t+        | Just t <- [jsonField "type" held]+        , t /= String (peerTypeName p.peerConfig)+        ]+            <> case p.peerConfig of+                PostgresSource c ->+                    fieldDiffs+                        (jsonField "postgresConfig" held)+                        [ ("host", String c.pgHost)+                        , ("port", Number (fromIntegral c.pgPort))+                        , ("user", String c.pgUser)+                        , ("database", String c.pgDatabase)+                        , ("requireTls", Bool c.pgRequireTls)+                        ]+                ClickHouseTarget c ->+                    fieldDiffs+                        (jsonField "clickhouseConfig" held)+                        [ ("host", String c.chHost)+                        , ("port", Number (fromIntegral c.chPort))+                        , ("user", String c.chUser)+                        , ("database", String c.chDatabase)+                        , ("disableTls", Bool c.chDisableTls)+                        ]+    fieldDiffs Nothing _ = ["not a " <> peerTypeName p.peerConfig <> " peer"]+    fieldDiffs (Just cfg) expected =+        [ k <> " is " <> maybe "unset" renderValue held <> ", declared " <> renderValue want+        | (k, want) <- expected+        , let held = jsonField k cfg+        , not (sameField want held)+        ]+    -- protobuf JSON leaves a false or empty field out+    sameField (Bool False) Nothing = True+    sameField (String "") Nothing = True+    sameField want held = Just want == held++-------------------------------------------------------------------------------+-- Mirrors++type MirrorName = Text++-- | A source table and the destination table it is mirrored to.+data TableMapping = TableMapping {sourceTable :: Text, destinationTable :: Text}+    deriving (Eq, Ord, Show)++{- | A CDC mirror from a Postgres peer to another peer.++'mirrorPublication' and 'mirrorSlot' name ones declared on the source+("Salmon.Builtin.Nodes.Postgres.Logical"); 'Nothing' lets PeerDB make its+own (it needs to own the tables for a publication, which a CDC role+usually does not).++A declared slot goes with no initial snapshot: PeerDB refuses an existing+slot for a snapshot (@irrecoverable_existing_slot@, the mirror goes to+@STATUS_FAILED@; seen at @stable-v0.37.12@), since the snapshot it copies+is the one it exports when it creates the slot itself. Such a mirror+streams what the slot holds from its creation on, and the rows already in+the tables are not copied. 'mirrorProblems' refuses the combination before+anything is sent.+-}+data Mirror+    = Mirror+    { mirrorName :: MirrorName+    , mirrorSource :: PeerName+    , mirrorDestination :: PeerName+    , mirrorTables :: [TableMapping]+    , mirrorPublication :: Maybe Text+    , mirrorSlot :: Maybe Text+    , mirrorInitialSnapshot :: Bool+    , mirrorMaxBatchSize :: Maybe Int+    , mirrorIdleTimeoutSeconds :: Maybe Int+    }+    deriving (Eq, Ord, Show)++-- | An initial snapshot, PeerDB's own publication and slot, its default batch size and timeout.+mirror :: MirrorName -> PeerName -> PeerName -> [TableMapping] -> Mirror+mirror name src dst tables =+    Mirror+        { mirrorName = name+        , mirrorSource = src+        , mirrorDestination = dst+        , mirrorTables = tables+        , mirrorPublication = Nothing+        , mirrorSlot = Nothing+        , mirrorInitialSnapshot = True+        , mirrorMaxBatchSize = Nothing+        , mirrorIdleTimeoutSeconds = Nothing+        }++{- | A CDC mirror, created when PeerDB has none of that name. It needs its+two peers: @inject@ them (or pass them through the 'Track'' 'Api').++@up@ asks for the mirror's status first, and:++* none: creates it (@POST \/v1\/flows\/cdc\/create@);+* one with the declared peers and tables, not failed: does nothing;+* one with other peers or tables, or one PeerDB reports failed or+  terminated: refuses (throws), and changes nothing. Changing a running+  mirror's tables, resyncing it, or dropping and recreating it are an+  operator's decisions.++The check is 'interpretMirrorStatus'. @down@ is nothing.+-}+mirrorOp :: Reporter Report -> Track' Api -> Api -> Mirror -> Op+mirrorOp r apiT api m =+    op "peerdb-mirror" (deps [run apiT api]) $ \actions ->+        actions+            { ref = mkRef "peerdb-mirror" (api.apiBaseUrl, m.mirrorName)+            , help = Text.unwords ["PeerDB CDC mirror", m.mirrorName, "from", m.mirrorSource, "to", m.mirrorDestination]+            , notes =+                ["flow api: " <> api.apiBaseUrl]+                    <> [t.sourceTable <> " -> " <> t.destinationTable | t <- sort m.mirrorTables]+                    <> ["publication " <> p | Just p <- [m.mirrorPublication]]+                    <> ["slot " <> s | Just s <- [m.mirrorSlot]]+                    <> ["no initial snapshot" | not m.mirrorInitialSnapshot]+            , check = checkMirror api m+            , up = ensureMirror r api m+            }++-- | What makes a declaration one PeerDB would accept and then fail, or refuse.+mirrorProblems :: Mirror -> [Text]+mirrorProblems m =+    ["no table is declared" | null m.mirrorTables]+        <> [ "slot " <> s <> " is declared with an initial snapshot, which PeerDB refuses on an existing slot: declare no snapshot, or no slot"+           | Just s <- [m.mirrorSlot]+           , m.mirrorInitialSnapshot+           ]++-- | The body of @POST \/v1\/flows\/cdc\/create@.+createMirrorBody :: Mirror -> Value+createMirrorBody m =+    object+        [ "connection_configs"+            .= object+                ( [ "flow_job_name" .= m.mirrorName+                  , "source_name" .= m.mirrorSource+                  , "destination_name" .= m.mirrorDestination+                  , "table_mappings"+                        .= [ object ["source_table_identifier" .= t.sourceTable, "destination_table_identifier" .= t.destinationTable]+                           | t <- m.mirrorTables+                           ]+                  , "do_initial_snapshot" .= m.mirrorInitialSnapshot+                  ]+                    <> ["publication_name" .= p | Just p <- [m.mirrorPublication]]+                    <> ["replication_slot_name" .= s | Just s <- [m.mirrorSlot]]+                    <> ["max_batch_size" .= n | Just n <- [m.mirrorMaxBatchSize]]+                    <> ["idle_timeout_seconds" .= n | Just n <- [m.mirrorIdleTimeoutSeconds]]+                )+        ]++-- | The body of @POST \/v1\/mirrors\/status@.+mirrorStatusBody :: MirrorName -> Value+mirrorStatusBody name = object ["flow_job_name" .= name, "include_flow_info" .= True, "exclude_batches" .= True]++-- | What PeerDB says about a mirror.+data MirrorState+    = NoSuchMirror+    | MirrorFound Value+    | MirrorUnreadable Text+    deriving (Show)++mirrorState :: Maybe (Reporter Report) -> Api -> MirrorName -> IO MirrorState+mirrorState r api name = do+    res <- case r of+        Just rep -> Right <$> call rep ("mirror " <> name) api "POST" "/v1/mirrors/status" (Just (mirrorStatusBody name))+        Nothing -> request api "POST" "/v1/mirrors/status" (Just (mirrorStatusBody name))+    pure $ case res of+        Left e -> MirrorUnreadable e+        Right (404, _) -> NoSuchMirror+        Right (200, body) -> maybe (MirrorUnreadable "an answer that is not JSON") MirrorFound (Aeson.decodeStrict body)+        Right (code, body) -> MirrorUnreadable (statusText code <> errorMessage (decode body))++checkMirror :: Api -> Mirror -> IO CheckResult+checkMirror api m = do+    st <- mirrorState Nothing api m.mirrorName+    pure $ case st of+        NoSuchMirror -> Failure ("PeerDB has no mirror " <> m.mirrorName)+        MirrorFound v -> interpretMirrorStatus m v+        MirrorUnreadable _ -> Unknown++ensureMirror :: Reporter Report -> Api -> Mirror -> IO ()+ensureMirror r api m = do+    st <- mirrorState (Just r) api m.mirrorName+    case st of+        MirrorUnreadable why -> throwIO (PeerDBRefused ("mirror " <> m.mirrorName <> ": cannot read its status: " <> why))+        MirrorFound v -> case interpretMirrorStatus m v of+            Failure why -> throwIO (PeerDBRefused (why <> "; salmon does not change, resync or recreate a mirror"))+            _ -> pure ()+        NoSuchMirror -> do+            case mirrorProblems m of+                [] -> pure ()+                problems -> throwIO (PeerDBRefused ("mirror " <> m.mirrorName <> ": " <> Text.intercalate "; " problems))+            (code, body) <- call r ("mirror " <> m.mirrorName) api "POST" "/v1/flows/cdc/create" (Just (createMirrorBody m))+            case code of+                200 -> pure ()+                _ -> throwIO (PeerDBRefused ("mirror " <> m.mirrorName <> ": " <> statusText code <> errorMessage (decode body)))++{- | A mirror's status read against its declaration:++* other peers, or another set of table mappings: 'Failure' naming them;+* @STATUS_FAILED@, @STATUS_TERMINATING@ or @STATUS_TERMINATED@: 'Failure';+* @STATUS_RUNNING@, @STATUS_PAUSED@ (an operator paused it: salmon does not+  resume it) or @STATUS_COMPLETED@: 'Success';+* anything else (setting up, snapshotting, pausing, modifying, resyncing,+  unknown): 'Unknown', it is on its way somewhere.+-}+interpretMirrorStatus :: Mirror -> Value -> CheckResult+interpretMirrorStatus m v =+    case (configDiffs, state) of+        (d : ds, _) -> Failure ("PeerDB mirror " <> m.mirrorName <> " differs: " <> Text.intercalate "; " (d : ds))+        ([], Just s)+            | s `elem` ["STATUS_FAILED", "STATUS_TERMINATING", "STATUS_TERMINATED"] -> Failure ("PeerDB mirror " <> m.mirrorName <> " is " <> s)+            | s `elem` ["STATUS_RUNNING", "STATUS_PAUSED", "STATUS_COMPLETED"] -> Success+        _ -> Unknown+  where+    state = case jsonField "currentFlowState" v of+        Just (String s) -> Just s+        _ -> Nothing+    config = jsonField "cdcStatus" v >>= jsonField "config"+    configDiffs = case config of+        Nothing -> ["not a CDC mirror"]+        Just c ->+            [ "source is " <> renderMaybe (jsonField "sourceName" c) | jsonField "sourceName" c /= Just (String m.mirrorSource)+            ]+                <> ["destination is " <> renderMaybe (jsonField "destinationName" c) | jsonField "destinationName" c /= Just (String m.mirrorDestination)]+                <> ["tables are " <> renderMappings held | let held = heldMappings c, held /= sort m.mirrorTables]+    heldMappings c = case jsonField "tableMappings" c of+        Just (Array xs) ->+            sort+                [ TableMapping s d+                | x <- toList xs+                , Just (String s) <- [jsonField "sourceTableIdentifier" x]+                , Just (String d) <- [jsonField "destinationTableIdentifier" x]+                ]+        _ -> []+    renderMappings :: [TableMapping] -> Text+    renderMappings [] = "none"+    renderMappings ts = Text.intercalate ", " [t.sourceTable <> " -> " <> t.destinationTable | t <- ts]+    renderMaybe = maybe "unset" renderValue++-------------------------------------------------------------------------------+-- HTTP++data PeerDBRefused = PeerDBRefused Text+    deriving (Show)++instance Exception PeerDBRefused++newManager' :: IO Manager+newManager' = newManager defaultManagerSettings{managerResponseTimeout = responseTimeoutMicro 60000000}++{- | One call to the flow API, reported (the method, path and status, never+the body). Throws a 'PeerDBRefused' when nothing answered.+-}+call :: Reporter Report -> Text -> Api -> Text -> Text -> Maybe Value -> IO (Int, ByteString.ByteString)+call r what api verb path body = do+    res <- request api verb path body+    case res of+        Left why -> do+            runReporter r (ApiCall what (verb <> " " <> path) 0)+            throwIO (PeerDBRefused (what <> ": the flow API at " <> api.apiBaseUrl <> " does not answer: " <> why))+        Right (code, out) -> do+            runReporter r (ApiCall what (verb <> " " <> path) code)+            pure (code, out)++-- | 'Left' when nothing answered; otherwise the status and the body.+request :: Api -> Text -> Text -> Maybe Value -> IO (Either Text (Int, ByteString.ByteString))+request api verb path body = do+    res <- try $ do+        mgr <- newManager'+        req0 <- parseRequest (Text.unpack (api.apiBaseUrl <> path))+        let req =+                req0+                    { method = Text.encodeUtf8 verb+                    , requestHeaders = [("Content-Type", "application/json") | Just _ <- [body]]+                    , requestBody = maybe (requestBody req0) (RequestBodyLBS . Aeson.encode) body+                    }+        rsp <- httpLbs req mgr+        pure (statusCode (responseStatus rsp), LByteString.toStrict (responseBody rsp))+    pure $ case res of+        -- the exception names the request (method, host, path), not its body+        Left (e :: HttpException) -> Left (Text.take 200 (httpExceptionText e))+        Right ok -> Right ok++httpExceptionText :: HttpException -> Text+httpExceptionText (Http.HttpExceptionRequest _ content) = Text.pack (show content)+httpExceptionText e = Text.pack (show e)++-- | A name as one URL path segment.+pathSegment :: Text -> Text+pathSegment = Text.decodeUtf8 . urlEncode False . Text.encodeUtf8++-- | The @message@ of a gRPC-gateway error, or the body as it is.+errorMessage :: Text -> Text+errorMessage body = case Aeson.decodeStrict (Text.encodeUtf8 body) >>= jsonField "message" of+    Just (String msg) -> ": " <> Text.take 400 msg+    _ -> if Text.null (Text.strip body) then "" else ": " <> Text.take 400 body++statusText :: Int -> Text+statusText code = "the flow API answered " <> showT code++redact :: [Text] -> Text -> Text+redact secrets t = foldr (\s acc -> if Text.null s then acc else Text.replace s "<redacted>" acc) t secrets++jsonField :: Text -> Value -> Maybe Value+jsonField k (Object o) = KeyMap.lookup (Key.fromText k) o+jsonField _ _ = Nothing++renderValue :: Value -> Text+renderValue (String s) = s+renderValue (Bool b) = if b then "true" else "false"+renderValue other = decode (LByteString.toStrict (Aeson.encode other))++showT :: (Show a) => a -> Text+showT = Text.pack . show++decode :: ByteString.ByteString -> Text+decode = Text.decodeUtf8With TextError.lenientDecode

@@ -0,0 +1,357 @@+{- | What a PostgreSQL cluster needs before something can read its changes+by logical decoding (a CDC tool such as PeerDB, see+"Salmon.Builtin.Nodes.PeerDB"): @wal_level = logical@, a role allowed to+replicate, a publication naming the tables, and possibly a logical+replication slot.++Every node here talks to the cluster as the @postgres@ OS user, as the rest+of "Salmon.Builtin.Nodes.Postgres" does, and has a real check.++Nothing here drops or lowers anything: @down@ is nothing for all four. A+publication or a slot that disappears under a running CDC mirror breaks it,+and PostgreSQL refuses to start with @wal_level@ below @logical@ while a+logical slot exists. Taking them away is an operator's decision.++None of these nodes names a primary: they are declared against one cluster+on one port, and on a standby they fail as PostgreSQL makes them fail.+-}+module Salmon.Builtin.Nodes.Postgres.Logical where++import Control.Exception (Exception, throwIO)+import Data.List (sort)+import Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.Encoding as Text++import Salmon.Actions.UpDown (CheckResult (..))+import Salmon.Builtin.Extension+import Salmon.Builtin.Nodes.Binary (Binary, withBinaryStdin)+import qualified Salmon.Builtin.Nodes.Binary as Binary+import Salmon.Builtin.Nodes.Postgres (+    DatabaseName,+    Port,+    PsqlBatch (..),+    ReplicationSlotName,+    RoleName,+    dollarQuote,+    psqlBatchIn_Sudo,+    psqlBatchRun_Sudo,+    psqlQueryIn_Sudo,+    psqlQuery_Sudo,+    quoteIdent,+    quoteLiteral,+ )+import Salmon.Op.Ref+import Salmon.Op.Track+import Salmon.Reporter++-------------------------------------------------------------------------------++data Report+    = LRWalLevel !Binary.Report+    | LRReplicationRole !RoleName !Binary.Report+    | LRPublication !PublicationName !Binary.Report+    | LRSlot !ReplicationSlotName !Binary.Report+    deriving (Show)++-------------------------------------------------------------------------------+-- wal_level++{- | @ALTER SYSTEM SET wal_level = logical@, then a reload, which applies+nothing (the setting needs a restart) but makes the server report the+setting as @pending_restart@: follow this node with+"Salmon.Builtin.Nodes.Postgres".'Salmon.Builtin.Nodes.Postgres.restartClusterIfPending'+(@restart \`inject\` logicalWalLevel ...@), which is the node that restarts.++The check reads what the configuration files say (the last entry of+@pg_file_settings@), and the running value when no file sets it: a level+written and waiting for its restart is this node done, the restart node's+business.++The 'Ref' is 'Salmon.Builtin.Nodes.Postgres.alterSystemSet''s for+@wal_level@ on this port, on purpose: it is the same effect site. A+declaration that also uses+'Salmon.Builtin.Nodes.Postgres.primaryReplicationSetup' (which sets+@replica@) has two nodes on one site, which the fold reports as+@Conflicting@ (last writer wins); @logical@ includes everything @replica@+gives a streaming standby, so drop the other one.++@down@ is nothing.+-}+logicalWalLevel :: Reporter Report -> Track' (Binary "psql") -> Port -> Op+logicalWalLevel r psql port =+    withBinaryStdin psql (psqlBatchRun_Sudo port) PsqlBatch (Text.encodeUtf8 walLevelSql) $ \apply ->+        op "pg-wal-level-logical" nodeps $ \actions ->+            actions+                { ref = mkRef "pg-alter-system" (port, "wal_level" :: Text)+                , help = "ALTER SYSTEM SET wal_level = logical"+                , notes = ["port " <> Text.pack (show port), "takes effect at the next restart of the cluster", "down leaves wal_level as it is"]+                , check = either (const Unknown) interpretWalLevel <$> psqlQuery_Sudo port walLevelInFilesSql+                , up = apply (contramap LRWalLevel r)+                }++walLevelSql :: Text+walLevelSql = "ALTER SYSTEM SET wal_level = 'logical';\nSELECT pg_reload_conf();\n"++-- | What the files say @wal_level@ is, or what the server runs with when no file sets it.+walLevelInFilesSql :: Text+walLevelInFilesSql =+    "SELECT coalesce((SELECT setting FROM pg_file_settings WHERE name = 'wal_level' ORDER BY seqno DESC LIMIT 1), current_setting('wal_level'))"++interpretWalLevel :: Text -> CheckResult+interpretWalLevel out = case Text.strip out of+    "logical" -> Success+    "" -> Unknown+    other -> Failure ("wal_level is " <> other <> ", logical decoding needs logical")++-------------------------------------------------------------------------------+-- The REPLICATION attribute++{- | @ALTER ROLE name WITH REPLICATION@ on a role declared elsewhere (a+'Salmon.Builtin.Nodes.Postgres.userPassFile', say: @inject@ it). Checked on+@pg_roles.rolreplication@. @down@ takes nothing away: another declaration+(a standby's replication user) may count on it.+-}+replicationRole :: Reporter Report -> Track' (Binary "psql") -> Port -> RoleName -> Op+replicationRole r psql port name =+    withBinaryStdin psql (psqlBatchRun_Sudo port) PsqlBatch (Text.encodeUtf8 (replicationRoleSql name)) $ \apply ->+        op "pg-role-replication" nodeps $ \actions ->+            actions+                { ref = mkRef "pg-role-replication" (port, name)+                , help = "give role " <> name <> " the REPLICATION attribute"+                , notes = ["port " <> Text.pack (show port)]+                , check = either (const Unknown) (interpretReplicationRole name) <$> psqlQuery_Sudo port (replicationRoleCheckSql name)+                , up = apply (contramap (LRReplicationRole name) r)+                }++replicationRoleSql :: RoleName -> Text+replicationRoleSql name = "ALTER ROLE " <> quoteIdent name <> " WITH REPLICATION;\n"++replicationRoleCheckSql :: RoleName -> Text+replicationRoleCheckSql name = "SELECT rolreplication FROM pg_roles WHERE rolname = " <> quoteLiteral name++interpretReplicationRole :: RoleName -> Text -> CheckResult+interpretReplicationRole name out = case Text.strip out of+    "t" -> Success+    "f" -> Failure ("role " <> name <> " may not replicate")+    "" -> Failure ("role " <> name <> " does not exist")+    _ -> Unknown++-------------------------------------------------------------------------------+-- Publications++type PublicationName = Text++-- | A table, by schema and name.+data QualifiedTable = QualifiedTable {tableSchema :: Text, tableName :: Text}+    deriving (Eq, Ord, Show)++-- | @public.name@.+publicTable :: Text -> QualifiedTable+publicTable = QualifiedTable "public"++renderQualified :: QualifiedTable -> Text+renderQualified t = t.tableSchema <> "." <> t.tableName++quoteQualified :: QualifiedTable -> Text+quoteQualified t = quoteIdent t.tableSchema <> "." <> quoteIdent t.tableName++-- | A publication of exactly these tables (at least one), in one database.+data Publication+    = Publication+    { pubName :: PublicationName+    , pubDatabase :: DatabaseName+    , pubTables :: [QualifiedTable]+    }+    deriving (Eq, Show)++-- | The comment prefix on a publication salmon made.+publicationMarker :: Text+publicationMarker = "salmon-publication:"++{- | A publication of the declared tables, created if missing and set to+exactly them if present (@ALTER PUBLICATION ... SET TABLE@: a table taken+out of the declaration stops being published; no row anywhere is touched).++Salmon only touches a publication it made: one it creates carries a+'publicationMarker' comment, and @up@ refuses (and changes nothing) when a+publication of that name exists without it. The whole batch is one+transaction.++Creating a publication of a table needs to own it; the @postgres@ OS user+does, which is the reason to declare the publication here rather than let a+CDC tool's less privileged role make its own.++The check: absent or unmarked is a 'Failure' (@up@ then creates or refuses),+a different set of tables is a 'Failure' naming them. @down@ is nothing.+-}+publication :: Reporter Report -> Track' (Binary "psql") -> Port -> Publication -> Op+publication r psql port pub =+    withBinaryStdin psql (psqlBatchIn_Sudo port pub.pubDatabase) PsqlBatch (Text.encodeUtf8 (publicationSql pub)) $ \apply ->+        op "pg-publication" nodeps $ \actions ->+            actions+                { ref = mkRef "pg-publication" (port, pub.pubDatabase, pub.pubName)+                , help = "publication " <> pub.pubName <> " in " <> pub.pubDatabase+                , notes = ["port " <> Text.pack (show port), "tables: " <> Text.intercalate ", " (fmap renderQualified (sort pub.pubTables))]+                , check = checkPublication port pub+                , up =+                    if null pub.pubTables+                        then throwIO (LogicalRefused ("publication " <> pub.pubName <> " declares no table"))+                        else apply (contramap (LRPublication pub.pubName) r)+                }++{- | The batch 'publication' sends: the refusal of an unmarked publication,+a @CREATE@ when there is none (through @\\gexec@), the @SET TABLE@ and the+marker, in a transaction.+-}+publicationSql :: Publication -> Text+publicationSql pub =+    Text.unlines+        [ "BEGIN;"+        , "DO "+            <> dollarQuote+                ( "BEGIN IF EXISTS (SELECT 1 FROM pg_publication p WHERE p.pubname = "+                    <> quoteLiteral pub.pubName+                    <> " AND coalesce(obj_description(p.oid, 'pg_publication'), '') NOT LIKE "+                    <> quoteLiteral (publicationMarker <> "%")+                    <> ") THEN RAISE EXCEPTION '%', "+                    <> quoteLiteral ("refusing to change publication " <> pub.pubName <> ": salmon did not make it (no " <> publicationMarker <> " comment)")+                    <> "; END IF; END"+                )+            <> ";"+        , "SELECT "+            <> quoteLiteral ("CREATE PUBLICATION " <> quoteIdent pub.pubName <> " FOR TABLE " <> tables)+            <> " WHERE NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = "+            <> quoteLiteral pub.pubName+            <> ") \\gexec"+        , "ALTER PUBLICATION " <> quoteIdent pub.pubName <> " SET TABLE " <> tables <> ";"+        , "COMMENT ON PUBLICATION " <> quoteIdent pub.pubName <> " IS " <> quoteLiteral (publicationMarker <> " " <> pub.pubDatabase) <> ";"+        , "COMMIT;"+        ]+  where+    tables = Text.intercalate ", " (fmap quoteQualified pub.pubTables)++-- | One row, @present:COMMENT@, when the publication exists.+publicationMarkerSql :: Publication -> Text+publicationMarkerSql pub =+    "SELECT 'present:' || coalesce(obj_description(oid, 'pg_publication'), '') FROM pg_publication WHERE pubname = "+        <> quoteLiteral pub.pubName++-- | The published tables, @schema.table@, one per line.+publicationTablesSql :: Publication -> Text+publicationTablesSql pub =+    "SELECT schemaname || '.' || tablename FROM pg_publication_tables WHERE pubname = "+        <> quoteLiteral pub.pubName+        <> " ORDER BY 1"++checkPublication :: Port -> Publication -> IO CheckResult+checkPublication port pub = do+    marker <- psqlQueryIn_Sudo port pub.pubDatabase (publicationMarkerSql pub)+    tables <- psqlQueryIn_Sudo port pub.pubDatabase (publicationTablesSql pub)+    pure $ case (marker, tables) of+        (Right m, Right t) -> interpretPublication pub m t+        _ -> Unknown++interpretPublication :: Publication -> Text -> Text -> CheckResult+interpretPublication pub markerOut tablesOut =+    case Text.stripPrefix "present:" (Text.strip markerOut) of+        Nothing -> Failure ("publication " <> pub.pubName <> " does not exist in " <> pub.pubDatabase)+        Just comment+            | not (publicationMarker `Text.isPrefixOf` comment) -> Failure ("publication " <> pub.pubName <> " exists but salmon did not make it")+            | held /= declared ->+                Failure+                    ( "publication "+                        <> pub.pubName+                        <> " publishes "+                        <> list held+                        <> ", declared "+                        <> list declared+                    )+            | otherwise -> Success+  where+    held = sort (filter (not . Text.null) (fmap Text.strip (Text.lines tablesOut)))+    declared = sort (fmap renderQualified pub.pubTables)+    list [] = "nothing"+    list xs = Text.intercalate ", " xs++-------------------------------------------------------------------------------+-- Logical replication slots++-- | The output plugin of the slots made here: the built-in one, which PeerDB and native subscriptions read.+slotPlugin :: Text+slotPlugin = "pgoutput"++{- | A logical replication slot (plugin @pgoutput@) in one database, created+if missing. A slot of that name that exists with another type, plugin or+database is refused, not replaced.++A slot keeps every WAL segment its consumer has not confirmed, for as long+as the consumer is away: a slot nobody reads fills the disk. Cap it with+@max_slot_wal_keep_size@ (past the cap PostgreSQL invalidates the slot and+the consumer has to start over). PeerDB makes its own slot for a mirror+that names none, which is the simpler choice; declare one here when it+should exist before the mirror, or outlive it.++The 'Ref' is the physical slot node's+('Salmon.Builtin.Nodes.Postgres.replicationSlot'), the slot's name: one+cluster has one slot of a name.++@down@ is nothing: dropping a slot under a mirror loses its place in the+stream.+-}+logicalSlot :: Reporter Report -> Track' (Binary "psql") -> Port -> DatabaseName -> ReplicationSlotName -> Op+logicalSlot r psql port db slot =+    withBinaryStdin psql (psqlBatchIn_Sudo port db) PsqlBatch (Text.encodeUtf8 (logicalSlotSql db slot)) $ \apply ->+        op "pg-logical-slot" nodeps $ \actions ->+            actions+                { ref = mkRef "pg-replication-slot" slot+                , help = "logical replication slot " <> slot <> " in " <> db+                , notes = ["port " <> Text.pack (show port), "plugin " <> slotPlugin, "down leaves the slot: an unread slot keeps WAL"]+                , check = either (const Unknown) (interpretSlot db slot) <$> psqlQueryIn_Sudo port db (slotCheckSql slot)+                , up = apply (contramap (LRSlot slot) r)+                }++logicalSlotSql :: DatabaseName -> ReplicationSlotName -> Text+logicalSlotSql db slot =+    Text.unlines+        [ "DO "+            <> dollarQuote+                ( "BEGIN IF EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = "+                    <> quoteLiteral slot+                    <> " AND NOT (slot_type = 'logical' AND plugin = "+                    <> quoteLiteral slotPlugin+                    <> " AND database = "+                    <> quoteLiteral db+                    <> ")) THEN RAISE EXCEPTION '%', "+                    <> quoteLiteral ("refusing slot " <> slot <> ": a slot of that name exists that is not a " <> slotPlugin <> " slot of " <> db)+                    <> "; END IF; END"+                )+            <> ";"+        , "SELECT pg_create_logical_replication_slot("+            <> quoteLiteral slot+            <> ", "+            <> quoteLiteral slotPlugin+            <> ") WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = "+            <> quoteLiteral slot+            <> ");"+        ]++-- | One row, @type|plugin|database@, when the slot exists.+slotCheckSql :: ReplicationSlotName -> Text+slotCheckSql slot =+    "SELECT slot_type || '|' || coalesce(plugin, '') || '|' || coalesce(database, '') FROM pg_replication_slots WHERE slot_name = "+        <> quoteLiteral slot++interpretSlot :: DatabaseName -> ReplicationSlotName -> Text -> CheckResult+interpretSlot db slot out = case Text.strip out of+    "" -> Failure ("replication slot " <> slot <> " does not exist")+    row+        | row == "logical|" <> slotPlugin <> "|" <> db -> Success+        | otherwise -> Failure ("replication slot " <> slot <> " is " <> row <> ", declared logical|" <> slotPlugin <> "|" <> db)++-------------------------------------------------------------------------------++newtype LogicalRefused = LogicalRefused Text+    deriving (Show)++instance Exception LogicalRefused