{-# LANGUAGE OverloadedLabels #-}
{-# LANGUAGE OverloadedStrings #-}

-- | Implements a Hydra network component using [etcd](https://etcd.io/).
--
-- While implementing a basic broadcast protocol over a distributed key-value
-- store is quite an overkill, the [Raft](https://raft.github.io/) consensus of
-- etcd provides our application with a crash-recovery fault-tolerant "atomic
-- broadcast" out-of-the-box. As a nice side-effect, the network layer becomes
-- very introspectable, while it would also support features like TLS or service
-- discovery.
--
-- The component installs, starts and configures an `etcd` instance and connects
-- to it using a GRPC client. We can only write and read from the cluster while
-- connected to the majority cluster.
--
-- Broadcasting is implemented using @put@ to some well-known key, while message
-- delivery is done by using @watch@ on the same 'msg' prefix. We keep a last known
-- revision, also stored on disk, to start 'watch' with that revision (+1) and
-- only deliver messages that were not seen before. In case we are not connected
-- to our 'etcd' instance or not enough peers (= on a minority cluster), we
-- retry sending, but also store messages to broadcast in a 'PersistentQueue',
-- which makes the node resilient against crashes while sending.
--
-- Connectivity and compatibility with other nodes on the cluster is tracked
-- using the key-value service as well:
--
--   * network connectivity is determined by being able to fetch the member list
--   * peer connectivity is tracked (best effort, not authorized) using an entry
--     at 'alive-\<advertise\>' keys with individual leases and repeated keep-alives
--   * each node compare-and-swaps its `version` into a key of the same name to
--     check compatibility (not updatable)
--
-- Note that the etcd cluster is configured to compact revisions down to 1000
-- every 5 minutes. This prevents infinite growth of the key-value store, but
-- also limits how long a node can be disconnected without missing out. 1000
-- should be more than enough for our use-case as the Hydra protocol will not
-- advance unless all participants are present.
module Hydra.Network.Etcd where

import Hydra.Prelude

import Cardano.Binary (decodeFull', serialize')
import Cardano.Crypto.Hash (SHA256, hashToStringAsHex, hashWithSerialiser)
import Codec.CBOR.Encoding qualified as CBOR
import Codec.CBOR.Write qualified as CBOR
import Control.Concurrent.Class.MonadSTM (
  isFullTBQueue,
  modifyTVar',
  peekTBQueue,
  readTBQueue,
  readTVarIO,
  swapTVar,
  tryPeekTBQueue,
  tryReadTBQueue,
  unGetTBQueue,
  writeTBQueue,
  writeTVar,
 )
import Control.Exception (IOException)
import Control.Lens ((^.), (^..), (^?))
import Data.Aeson (decodeFileStrict', encodeFile)
import Data.Aeson qualified as Aeson
import Data.Aeson.Lens qualified as Aeson
import Data.Aeson.Types (Value)
import Data.ByteString qualified as BS
import Data.ByteString.Char8 qualified as BS8
import Data.List ((\\))
import Data.List qualified as List
import Data.Map.Strict qualified as Map
import Data.Text qualified as T
import Hydra.Logging (Tracer, traceWith)
import Hydra.Network (
  Connectivity (..),
  Host (..),
  Network (..),
  NetworkCallback (..),
  NetworkComponent,
  NetworkConfiguration (..),
  ProtocolVersion,
 )
import Hydra.Network.EtcdBinary (getEtcdBinary)
import Network.GRPC.Client (
  Address (..),
  ConnParams (..),
  Connection,
  ReconnectPolicy (..),
  ReconnectTo (ReconnectToOriginal),
  Server (..),
  ServerDisconnected,
  Timeout (..),
  TimeoutUnit (..),
  TimeoutValue (..),
  rpc,
  withConnection,
 )
import Network.GRPC.Client.StreamType.IO (biDiStreaming, nonStreaming)
import Network.GRPC.Common (GrpcError (..), GrpcException (..), HTTP2Settings (..), NextElem (..), def, defaultHTTP2Settings)
import Network.GRPC.Common.Protobuf (Proto (..), Protobuf, defMessage, (.~))
import Network.GRPC.Etcd (
  Compare'CompareResult (..),
  Compare'CompareTarget (..),
  KV,
  Lease,
  Watch,
 )
import Network.HTTP2.Client (HTTP2Error)
import Network.Socket (PortNumber)
import System.Directory (createDirectoryIfMissing, listDirectory, removeFile, renameFile)
import System.Environment.Blank (getEnvironment)
import System.FilePath ((</>))
import System.IO.Error (isDoesNotExistError, isEOFError)
import System.Process (interruptProcessGroupOf)
import System.Process.Typed (
  Process,
  ProcessConfig,
  createPipe,
  getStderr,
  proc,
  setCreateGroup,
  setEnv,
  setStderr,
  startProcess,
  stopProcess,
  unsafeProcessHandle,
  waitExitCode,
 )

-- | Concrete network component that broadcasts messages to an etcd cluster and
-- listens for incoming messages.
withEtcdNetwork ::
  (ToCBOR msg, FromCBOR msg) =>
  Tracer IO EtcdLog ->
  ProtocolVersion ->
  -- TODO: check if all of these needed?
  NetworkConfiguration ->
  NetworkComponent IO msg msg ()
withEtcdNetwork :: forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork Tracer IO EtcdLog
tracer ProtocolVersion
protocolVersion NetworkConfiguration
config NetworkCallback msg IO
callback Network IO msg -> IO ()
action = do
  String
etcdBinPath <- String -> WhichEtcd -> IO String
getEtcdBinary String
persistenceDir WhichEtcd
whichEtcd
  -- TODO: fail if cluster config / members do not match --peer
  -- configuration? That would be similar to the 'acks' persistence
  -- bailing out on loading.
  Map String String
envVars <- [(String, String)] -> Map String String
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(String, String)] -> Map String String)
-> IO [(String, String)] -> IO (Map String String)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO [(String, String)]
getEnvironment
  -- Bounded buffer of recent etcd stderr lines; rendered into the failure
  -- message below, which is otherwise silent about the reason etcd exited
  -- (e.g. which port was already bound).
  IORef [Text]
recentStderr <- [Text] -> IO (IORef [Text])
forall (m :: * -> *) a. MonadIO m => a -> m (IORef a)
newIORef []
  ProcessConfig () () Handle
-> (Process () () Handle -> IO ()) -> IO ()
forall (m :: * -> *) stdin stdout stderr a.
(MonadIO m, MonadThrow m) =>
ProcessConfig stdin stdout stderr
-> (Process stdin stdout stderr -> m a) -> m a
withProcessInterrupt (String -> Map String String -> ProcessConfig () () Handle
etcdCmd String
etcdBinPath Map String String
envVars) ((Process () () Handle -> IO ()) -> IO ())
-> (Process () () Handle -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Process () () Handle
p -> do
    (String, IO Any) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
      ( String
"etcd-waitExitCode"
      , do
          ExitCode
ec <- Process () () Handle -> IO ExitCode
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
Process stdin stdout stderr -> m ExitCode
waitExitCode Process () () Handle
p
          -- Let 'traceStderr' drain the last lines out of the closing pipe.
          DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
0.1
          [Text]
stderrLines <- [Text] -> [Text]
forall a. [a] -> [a]
reverse ([Text] -> [Text]) -> IO [Text] -> IO [Text]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef [Text] -> IO [Text]
forall (m :: * -> *) a. MonadIO m => IORef a -> m a
readIORef IORef [Text]
recentStderr
          String -> IO Any
forall a. String -> IO a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail (String -> IO Any) -> ([Text] -> String) -> [Text] -> IO Any
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> String
forall a. ToString a => a -> String
toString (Text -> String) -> ([Text] -> Text) -> [Text] -> String
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Text] -> Text
forall t. IsText t "unlines" => [t] -> t
unlines ([Text] -> IO Any) -> [Text] -> IO Any
forall a b. (a -> b) -> a -> b
$
            (Text
"Sub-process etcd exited with: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> ExitCode -> Text
forall b a. (Show a, IsString b) => a -> b
show ExitCode
ec)
              Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: if [Text] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Text]
stderrLines then [] else Text
"Recent etcd output:" Text -> [Text] -> [Text]
forall a. a -> [a] -> [a]
: [Text]
stderrLines
      )
      ( String
"etcd-callback-1"
      , (String, IO Any) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
          (String
"etcd-traceStderr", IORef [Text]
-> Process () () Handle -> NetworkCallback msg IO -> IO Any
traceStderr IORef [Text]
recentStderr Process () () Handle
p NetworkCallback msg IO
callback)
          ( String
"etcd-callback-2"
          , do
              -- NOTE: The connection to the server is set up asynchronously; the
              -- first rpc call will block until the connection has been established.
              ConnParams -> Server -> (Connection -> IO ()) -> IO ()
forall a. ConnParams -> Server -> (Connection -> IO a) -> IO a
withConnection (Tracer IO EtcdLog -> Maybe Timeout -> ConnParams
connParams Tracer IO EtcdLog
tracer Maybe Timeout
forall a. Maybe a
Nothing) (NetworkConfiguration -> Server
grpcServer NetworkConfiguration
config) ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
                -- REVIEW: checkVersion blocks if used on main thread - why?
                (String, IO ()) -> (Async IO () -> IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (Async m a -> m b) -> m b
withAsyncLabelled (String
"etcd-checkVersion", Tracer IO EtcdLog
-> Connection -> ProtocolVersion -> NetworkCallback msg IO -> IO ()
forall msg.
Tracer IO EtcdLog
-> Connection -> ProtocolVersion -> NetworkCallback msg IO -> IO ()
checkVersion Tracer IO EtcdLog
tracer Connection
conn ProtocolVersion
protocolVersion NetworkCallback msg IO
callback) ((Async IO () -> IO ()) -> IO ())
-> (Async IO () -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Async IO ()
_ -> do
                  (String, IO ()) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
                    (String
"etcd-pollConnectivity", Tracer IO EtcdLog
-> Connection -> Host -> NetworkCallback msg IO -> IO ()
forall msg.
Tracer IO EtcdLog
-> Connection -> Host -> NetworkCallback msg IO -> IO ()
pollConnectivity Tracer IO EtcdLog
tracer Connection
conn Host
advertise NetworkCallback msg IO
callback)
                    ( String
"etcd-callback-3"
                    , (String, IO ()) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
                        (String
"etcd-waitMessages", Tracer IO EtcdLog
-> Connection -> String -> NetworkCallback msg IO -> IO ()
forall msg.
FromCBOR msg =>
Tracer IO EtcdLog
-> Connection -> String -> NetworkCallback msg IO -> IO ()
waitMessages Tracer IO EtcdLog
tracer Connection
conn String
persistenceDir NetworkCallback msg IO
callback)
                        ( String
"etcd-callback-4"
                        , do
                            PersistentQueue IO msg
queue <- Tracer IO EtcdLog
-> String -> Natural -> IO (PersistentQueue IO msg)
forall (m :: * -> *) a.
(MonadLabelledSTM m, MonadIO m, FromCBOR a, MonadCatch m,
 MonadFail m) =>
Tracer IO EtcdLog -> String -> Natural -> m (PersistentQueue m a)
newPersistentQueue Tracer IO EtcdLog
tracer (String
persistenceDir String -> String -> String
</> String
"pending-broadcast") Natural
100
                            (String, IO ()) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
                              (String
"etcd-broadcastMessages", Tracer IO EtcdLog
-> NetworkConfiguration -> Host -> PersistentQueue IO msg -> IO ()
forall msg.
Tracer IO EtcdLog
-> NetworkConfiguration -> Host -> PersistentQueue IO msg -> IO ()
broadcastMessages Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
advertise PersistentQueue IO msg
queue)
                              ( String
"etcd-network-component-action"
                              , do
                                  Network IO msg -> IO ()
action
                                    Network
                                      { $sel:broadcast:Network :: msg -> IO ()
broadcast = Tracer IO EtcdLog -> PersistentQueue IO msg -> msg -> IO ()
forall a (m :: * -> *).
(ToCBOR a, MonadSTM m, MonadIO m) =>
Tracer IO EtcdLog -> PersistentQueue m a -> a -> m ()
writePersistentQueue Tracer IO EtcdLog
tracer PersistentQueue IO msg
queue
                                      }
                              )
                        )
                    )
          )
      )
 where
  clientHost :: Host
clientHost = Host{$sel:hostname:Host :: Text
hostname = Text
"127.0.0.1", $sel:port:Host :: PortNumber
port = NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration
config}
  traceStderr :: IORef [Text]
-> Process () () Handle -> NetworkCallback msg IO -> IO Any
traceStderr IORef [Text]
recentStderr Process () () Handle
p NetworkCallback{Connectivity -> IO ()
onConnectivity :: Connectivity -> IO ()
$sel:onConnectivity:NetworkCallback :: forall msg (m :: * -> *).
NetworkCallback msg m -> Connectivity -> m ()
onConnectivity} =
    let loop :: IO Any
loop = do
          ByteString
bs <- Handle -> IO ByteString
BS.hGetLine (Process () () Handle -> Handle
forall stdin stdout stderr. Process stdin stdout stderr -> stderr
getStderr Process () () Handle
p)
          IORef [Text] -> ([Text] -> [Text]) -> IO ()
forall (m :: * -> *) a. MonadIO m => IORef a -> (a -> a) -> m ()
modifyIORef' IORef [Text]
recentStderr (Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
take Int
20 ([Text] -> [Text]) -> ([Text] -> [Text]) -> [Text] -> [Text]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 ByteString
bs :))
          case ByteString -> Either String Value
forall a. FromJSON a => ByteString -> Either String a
Aeson.eitherDecodeStrict ByteString
bs of
            Left String
err -> Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer FailedToDecodeLog{$sel:log:EtcdLog :: Text
log = ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 ByteString
bs, $sel:reason:EtcdLog :: Text
reason = String -> Text
forall b a. (Show a, IsString b) => a -> b
show String
err}
            Right Value
v -> do
              let expectedClusterMismatch :: Maybe (Value, Value)
expectedClusterMismatch = do
                    Value
level' <- ByteString
bs ByteString -> Getting (First Value) ByteString Value -> Maybe Value
forall s a. s -> Getting (First a) s a -> Maybe a
^? Key -> Traversal' ByteString Value
forall t. AsValue t => Key -> Traversal' t Value
Aeson.key Key
"level" Getting (First Value) ByteString Value
-> ((Value -> Const (First Value) Value)
    -> Value -> Const (First Value) Value)
-> Getting (First Value) ByteString Value
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Value -> Const (First Value) Value)
-> Value -> Const (First Value) Value
Prism' Value Value
Aeson.nonNull
                    Value
msg' <- ByteString
bs ByteString -> Getting (First Value) ByteString Value -> Maybe Value
forall s a. s -> Getting (First a) s a -> Maybe a
^? Key -> Traversal' ByteString Value
forall t. AsValue t => Key -> Traversal' t Value
Aeson.key Key
"msg" Getting (First Value) ByteString Value
-> ((Value -> Const (First Value) Value)
    -> Value -> Const (First Value) Value)
-> Getting (First Value) ByteString Value
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Value -> Const (First Value) Value)
-> Value -> Const (First Value) Value
Prism' Value Value
Aeson.nonNull
                    (Value, Value) -> Maybe (Value, Value)
forall a. a -> Maybe a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Value
level', Value
msg')
              case Maybe (Value, Value)
expectedClusterMismatch of
                Just (Aeson.String Text
"error", Aeson.String Text
"request sent was ignored due to cluster ID mismatch") ->
                  Connectivity -> IO ()
onConnectivity ClusterIDMismatch{$sel:clusterPeers:PeerConnected :: Text
clusterPeers = String -> Text
T.pack String
clusterPeers}
                Maybe (Value, Value)
_ -> Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer (EtcdLog -> IO ()) -> EtcdLog -> IO ()
forall a b. (a -> b) -> a -> b
$ EtcdLog{etcd :: Value
etcd = Value
v}
          IO Any
loop
     in -- When etcd's stderr pipe closes (because the etcd subprocess exited),
        -- 'BS.hGetLine' raises 'hGetLine: end of file'. We don't want that
        -- naked IOException racing the descriptive "Sub-process etcd exited
        -- with: ExitFailure N" from 'etcd-waitExitCode' below — on slower
        -- machines the EOF often wins, and the IOException then gets caught
        -- by 'withAPIServer's IOException handler and re-thrown as
        -- 'RunServerException', stripping every mention of etcd from the
        -- final error. Block on EOF instead so 'etcd-waitExitCode' is always
        -- the one that fires.
        IO Any
loop IO Any -> (IOError -> IO Any) -> IO Any
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \IOError
e ->
          if IOError -> Bool
isEOFError IOError
e
            then IO () -> IO Any
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
60)
            else IOError -> IO Any
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO IOError
e

  -- XXX: Could use TLS to secure peer connections
  -- XXX: Could use discovery to simplify configuration
  -- NOTE: Configured using guides: https://etcd.io/docs/v3.5/op-guide
  etcdCmd :: String -> Map String String -> ProcessConfig () () Handle
etcdCmd String
etcdBinPath Map String String
envVars =
    -- NOTE: We map prefers the left; so we need to mappend default at the end.
    [(String, String)]
-> ProcessConfig () () Handle -> ProcessConfig () () Handle
forall stdin stdout stderr.
[(String, String)]
-> ProcessConfig stdin stdout stderr
-> ProcessConfig stdin stdout stderr
setEnv (Map String String -> [(String, String)]
forall k a. Map k a -> [(k, a)]
Map.toList (Map String String -> [(String, String)])
-> Map String String -> [(String, String)]
forall a b. (a -> b) -> a -> b
$ Map String String
envVars Map String String -> Map String String -> Map String String
forall a. Semigroup a => a -> a -> a
<> Map String String
defaultEnv)
      (ProcessConfig () () Handle -> ProcessConfig () () Handle)
-> ([String] -> ProcessConfig () () Handle)
-> [String]
-> ProcessConfig () () Handle
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Bool -> ProcessConfig () () Handle -> ProcessConfig () () Handle
forall stdin stdout stderr.
Bool
-> ProcessConfig stdin stdout stderr
-> ProcessConfig stdin stdout stderr
setCreateGroup Bool
True -- Prevents interrupt of main process when we send SIGINT to etcd
      (ProcessConfig () () Handle -> ProcessConfig () () Handle)
-> ([String] -> ProcessConfig () () Handle)
-> [String]
-> ProcessConfig () () Handle
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StreamSpec 'STOutput Handle
-> ProcessConfig () () () -> ProcessConfig () () Handle
forall stderr stdin stdout stderr0.
StreamSpec 'STOutput stderr
-> ProcessConfig stdin stdout stderr0
-> ProcessConfig stdin stdout stderr
setStderr StreamSpec 'STOutput Handle
forall (anyStreamType :: StreamType).
StreamSpec anyStreamType Handle
createPipe
      (ProcessConfig () () () -> ProcessConfig () () Handle)
-> ([String] -> ProcessConfig () () ())
-> [String]
-> ProcessConfig () () Handle
forall b c a. (b -> c) -> (a -> b) -> a -> c
. String -> [String] -> ProcessConfig () () ()
proc String
etcdBinPath
      ([String] -> ProcessConfig () () Handle)
-> [String] -> ProcessConfig () () Handle
forall a b. (a -> b) -> a -> b
$ [[String]] -> [String]
forall (t :: * -> *) a. Foldable t => t [a] -> [a]
concat
        [ -- NOTE: Must be used in clusterPeers
          [String
"--name", Host -> String
forall b a. (Show a, IsString b) => a -> b
show Host
advertise]
        , [String
"--data-dir", String
persistenceDir String -> String -> String
</> String
"etcd" String -> String -> String
</> Hash SHA256 ByteString -> String
forall h a. Hash h a -> String
hashToStringAsHex (forall h a. HashAlgorithm h => (a -> Encoding) -> a -> Hash h a
hashWithSerialiser @SHA256 ByteString -> Encoding
forall a. ToCBOR a => a -> Encoding
toCBOR (ByteString -> Hash SHA256 ByteString)
-> ByteString -> Hash SHA256 ByteString
forall a b. (a -> b) -> a -> b
$ String -> ByteString
BS8.pack String
clusterPeers)]
        , [String
"--listen-peer-urls", Host -> String
httpUrl Host
listen]
        , [String
"--initial-advertise-peer-urls", Host -> String
httpUrl Host
advertise]
        , [String
"--listen-client-urls", Host -> String
httpUrl Host
clientHost]
        , -- Pick a random port for http api (and use above only for grpc)
          [String
"--listen-client-http-urls", String
"http://localhost:0"]
        , -- Client access only on configured 'host' interface.
          [String
"--advertise-client-urls", Host -> String
httpUrl Host
clientHost]
        , -- XXX: Could use unique initial-cluster-tokens to isolate clusters
          [String
"--initial-cluster-token", String
"hydra-network-1"]
        , [String
"--initial-cluster", String
clusterPeers]
        ]

  defaultEnv :: Map.Map String String
  defaultEnv :: Map String String
defaultEnv =
    -- Keep up to 1000 revisions. See also:
    -- https://etcd.io/docs/v3.5/op-guide/maintenance/#auto-compaction
    [(String, String)] -> Map String String
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList
      [ (String
"ETCD_AUTO_COMPACTION_MODE", String
"revision")
      , (String
"ETCD_AUTO_COMPACTION_RETENTION", String
"1000")
      ]

  -- NOTE: Building a canonical list of labels from the advertised addresses
  clusterPeers :: String
clusterPeers =
    String -> [String] -> String
forall a. [a] -> [[a]] -> [a]
intercalate String
","
      ([String] -> String) -> ([Host] -> [String]) -> [Host] -> String
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Host -> String) -> [Host] -> [String]
forall a b. (a -> b) -> [a] -> [b]
map (\Host
h -> Host -> String
forall b a. (Show a, IsString b) => a -> b
show Host
h String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"=" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Host -> String
httpUrl Host
h)
      ([Host] -> String) -> [Host] -> String
forall a b. (a -> b) -> a -> b
$ (Host
advertise Host -> [Host] -> [Host]
forall a. a -> [a] -> [a]
: [Host]
peers)

  httpUrl :: Host -> String
httpUrl (Host Text
h PortNumber
p) = String
"http://" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Text -> String
forall a. ToString a => a -> String
toString Text
h String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
":" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> PortNumber -> String
forall b a. (Show a, IsString b) => a -> b
show PortNumber
p

  NetworkConfiguration{String
persistenceDir :: String
$sel:persistenceDir:NetworkConfiguration :: NetworkConfiguration -> String
persistenceDir, Host
listen :: Host
$sel:listen:NetworkConfiguration :: NetworkConfiguration -> Host
listen, Host
advertise :: Host
$sel:advertise:NetworkConfiguration :: NetworkConfiguration -> Host
advertise, [Host]
peers :: [Host]
$sel:peers:NetworkConfiguration :: NetworkConfiguration -> [Host]
peers, WhichEtcd
whichEtcd :: WhichEtcd
$sel:whichEtcd:NetworkConfiguration :: NetworkConfiguration -> WhichEtcd
whichEtcd} = NetworkConfiguration
config

connParams :: Tracer IO EtcdLog -> Maybe Timeout -> ConnParams
connParams :: Tracer IO EtcdLog -> Maybe Timeout -> ConnParams
connParams Tracer IO EtcdLog
tracer Maybe Timeout
to =
  ConnParams
forall a. Default a => a
def
    { connReconnectPolicy = reconnectPolicy
    , -- NOTE: Do not rate limit pings or settings from our trusted, local etcd
      -- node; see the comments on both override fields. The two guards trip on
      -- the same thing: grpc-go raises its receive window as inbound volume
      -- grows, pinging to measure the round-trip and sending a SETTINGS frame
      -- per step. Sustained broadcast walks more steps than the default
      -- 4 SETTINGS/s allows, and http2 then kills the connection (#2817).
      connHTTP2Settings =
        defaultHTTP2Settings
          { http2OverridePingRateLimit = Just maxBound
          , http2OverrideSettingsRateLimit = Just maxBound
          }
    , connDefaultTimeout = to
    }
 where
  reconnectPolicy :: ReconnectPolicy
reconnectPolicy = ReconnectTo -> IO ReconnectPolicy -> ReconnectPolicy
ReconnectAfter ReconnectTo
ReconnectToOriginal (IO ReconnectPolicy -> ReconnectPolicy)
-> IO ReconnectPolicy -> ReconnectPolicy
forall a b. (a -> b) -> a -> b
$ do
    DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1
    Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer EtcdLog
Reconnecting
    ReconnectPolicy -> IO ReconnectPolicy
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ReconnectPolicy
reconnectPolicy

grpcServer :: NetworkConfiguration -> Server
grpcServer :: NetworkConfiguration -> Server
grpcServer NetworkConfiguration
config =
  Address -> Server
ServerInsecure (Address -> Server) -> Address -> Server
forall a b. (a -> b) -> a -> b
$
    Address
      { addressHost :: String
addressHost = Text -> String
forall a. ToString a => a -> String
toString (Text -> String) -> Text -> String
forall a b. (a -> b) -> a -> b
$ Host -> Text
hostname Host
clientHost
      , addressPort :: PortNumber
addressPort = Host -> PortNumber
port Host
clientHost
      , addressAuthority :: Maybe String
addressAuthority = Maybe String
forall a. Maybe a
Nothing
      }
 where
  clientHost :: Host
clientHost = Host{$sel:hostname:Host :: Text
hostname = Text
"127.0.0.1", $sel:port:Host :: PortNumber
port = NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration
config}

-- | Get the client port corresponding to a listen address.
--
-- The client port used by the started etcd port is offset by the same amount as
-- the listen address is offset by the default port 5001. This will result in
-- the default client port 2379 be used by default still.
getClientPort :: NetworkConfiguration -> PortNumber
getClientPort :: NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration{Host
$sel:listen:NetworkConfiguration :: NetworkConfiguration -> Host
listen :: Host
listen} = PortNumber -> PortNumber
peerPortToClientPort (Host -> PortNumber
port Host
listen)

-- | Derive the etcd client port from a configured peer (listen) port.
--
-- Exposed separately from 'getClientPort' so test fixtures can pre-allocate
-- both the peer and the derived client port without constructing a full
-- 'NetworkConfiguration'. Keep this and 'getClientPort' in lockstep — any
-- change to the offset must happen here, in one place.
peerPortToClientPort :: PortNumber -> PortNumber
peerPortToClientPort :: PortNumber -> PortNumber
peerPortToClientPort PortNumber
listenPort = PortNumber
2379 PortNumber -> PortNumber -> PortNumber
forall a. Num a => a -> a -> a
+ PortNumber
listenPort PortNumber -> PortNumber -> PortNumber
forall a. Num a => a -> a -> a
- PortNumber
5001

-- | Check and write version on etcd cluster. This will retry until we are on a
-- majority cluster and succeed. If the version does not match a corresponding
-- 'Connectivity' message is sent via 'NetworkCallback'.
checkVersion ::
  Tracer IO EtcdLog ->
  Connection ->
  ProtocolVersion ->
  NetworkCallback msg IO ->
  IO ()
checkVersion :: forall msg.
Tracer IO EtcdLog
-> Connection -> ProtocolVersion -> NetworkCallback msg IO -> IO ()
checkVersion Tracer IO EtcdLog
tracer Connection
conn ProtocolVersion
ourVersion NetworkCallback{Connectivity -> IO ()
$sel:onConnectivity:NetworkCallback :: forall msg (m :: * -> *).
NetworkCallback msg m -> Connectivity -> m ()
onConnectivity :: Connectivity -> IO ()
onConnectivity} = do
  -- Get or write our version into kv store
  Proto TxnResponse
res <-
    Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf KV "txn")
-> Input (Protobuf KV "txn")
-> IO (Output (Protobuf KV "txn"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf KV "txn")) (Input (Protobuf KV "txn") -> IO (Output (Protobuf KV "txn")))
-> Input (Protobuf KV "txn") -> IO (Output (Protobuf KV "txn"))
forall a b. (a -> b) -> a -> b
$
      Proto TxnRequest
forall msg. Message msg => msg
defMessage
        Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto Compare]
  [Proto Compare]
#compare ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto Compare]
  [Proto Compare]
-> [Proto Compare] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Proto Compare
versionExists]
        Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
#success ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
-> [Proto RequestOp] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Proto RequestOp
getVersion]
        Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
#failure ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
-> [Proto RequestOp] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Proto RequestOp
putVersion]

  -- Check version if version was already present
  if Proto TxnResponse
res Proto TxnResponse -> Getting Bool (Proto TxnResponse) Bool -> Bool
forall s a. s -> Getting a s a -> a
^. Getting Bool (Proto TxnResponse) Bool
#succeeded
    then [Proto KeyValue] -> (Proto KeyValue -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (Proto TxnResponse
res Proto TxnResponse
-> Getting
     (Endo [Proto KeyValue]) (Proto TxnResponse) (Proto KeyValue)
-> [Proto KeyValue]
forall s a. s -> Getting (Endo [a]) s a -> [a]
^.. ([Proto ResponseOp]
 -> Const (Endo [Proto KeyValue]) [Proto ResponseOp])
-> Proto TxnResponse
-> Const (Endo [Proto KeyValue]) (Proto TxnResponse)
#responses (([Proto ResponseOp]
  -> Const (Endo [Proto KeyValue]) [Proto ResponseOp])
 -> Proto TxnResponse
 -> Const (Endo [Proto KeyValue]) (Proto TxnResponse))
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> [Proto ResponseOp]
    -> Const (Endo [Proto KeyValue]) [Proto ResponseOp])
-> Getting
     (Endo [Proto KeyValue]) (Proto TxnResponse) (Proto KeyValue)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto ResponseOp
 -> Const (Endo [Proto KeyValue]) (Proto ResponseOp))
-> [Proto ResponseOp]
-> Const (Endo [Proto KeyValue]) [Proto ResponseOp]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse ((Proto ResponseOp
  -> Const (Endo [Proto KeyValue]) (Proto ResponseOp))
 -> [Proto ResponseOp]
 -> Const (Endo [Proto KeyValue]) [Proto ResponseOp])
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> Proto ResponseOp
    -> Const (Endo [Proto KeyValue]) (Proto ResponseOp))
-> (Proto KeyValue
    -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> [Proto ResponseOp]
-> Const (Endo [Proto KeyValue]) [Proto ResponseOp]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto RangeResponse
 -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
-> Proto ResponseOp
-> Const (Endo [Proto KeyValue]) (Proto ResponseOp)
#responseRange ((Proto RangeResponse
  -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
 -> Proto ResponseOp
 -> Const (Endo [Proto KeyValue]) (Proto ResponseOp))
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> Proto RangeResponse
    -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
-> (Proto KeyValue
    -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> Proto ResponseOp
-> Const (Endo [Proto KeyValue]) (Proto ResponseOp)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ([Proto KeyValue]
 -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
-> Proto RangeResponse
-> Const (Endo [Proto KeyValue]) (Proto RangeResponse)
#kvs (([Proto KeyValue]
  -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
 -> Proto RangeResponse
 -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> [Proto KeyValue]
    -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
-> (Proto KeyValue
    -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> Proto RangeResponse
-> Const (Endo [Proto KeyValue]) (Proto RangeResponse)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto KeyValue -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> [Proto KeyValue]
-> Const (Endo [Proto KeyValue]) [Proto KeyValue]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse) ((Proto KeyValue -> IO ()) -> IO ())
-> (Proto KeyValue -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Proto KeyValue
kv ->
      case ByteString -> Either DecoderError ProtocolVersion
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' (ByteString -> Either DecoderError ProtocolVersion)
-> ByteString -> Either DecoderError ProtocolVersion
forall a b. (a -> b) -> a -> b
$ Proto KeyValue
kv Proto KeyValue
-> Getting ByteString (Proto KeyValue) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto KeyValue) ByteString
#value of
        Left DecoderError
err -> do
          Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer (EtcdLog -> IO ()) -> EtcdLog -> IO ()
forall a b. (a -> b) -> a -> b
$
            FailedToDecodeValue
              { $sel:key:EtcdLog :: Text
key = ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 (ByteString -> Text) -> ByteString -> Text
forall a b. (a -> b) -> a -> b
$ Proto KeyValue
kv Proto KeyValue
-> Getting ByteString (Proto KeyValue) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto KeyValue) ByteString
#key
              , $sel:value:EtcdLog :: Text
value = ByteString -> Text
encodeBase16 (ByteString -> Text) -> ByteString -> Text
forall a b. (a -> b) -> a -> b
$ Proto KeyValue
kv Proto KeyValue
-> Getting ByteString (Proto KeyValue) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto KeyValue) ByteString
#value
              , $sel:reason:EtcdLog :: Text
reason = DecoderError -> Text
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err
              }
          Connectivity -> IO ()
onConnectivity VersionMismatch{ProtocolVersion
ourVersion :: ProtocolVersion
$sel:ourVersion:PeerConnected :: ProtocolVersion
ourVersion, $sel:theirVersion:PeerConnected :: Maybe ProtocolVersion
theirVersion = Maybe ProtocolVersion
forall a. Maybe a
Nothing}
        Right ProtocolVersion
theirVersion ->
          Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (ProtocolVersion
theirVersion ProtocolVersion -> ProtocolVersion -> Bool
forall a. Eq a => a -> a -> Bool
== ProtocolVersion
ourVersion) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connectivity -> IO ()
onConnectivity VersionMismatch{ProtocolVersion
ourVersion :: ProtocolVersion
$sel:ourVersion:PeerConnected :: ProtocolVersion
ourVersion, $sel:theirVersion:PeerConnected :: Maybe ProtocolVersion
theirVersion = ProtocolVersion -> Maybe ProtocolVersion
forall a. a -> Maybe a
Just ProtocolVersion
theirVersion}
    else Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer (EtcdLog -> IO ()) -> EtcdLog -> IO ()
forall a b. (a -> b) -> a -> b
$ MatchingProtocolVersion{version :: ProtocolVersion
version = ProtocolVersion
ourVersion}
 where
  versionKey :: ByteString
versionKey = ByteString
"version"

  -- exists = create_revision of key 'version' > 0
  versionExists :: Proto Compare
versionExists =
    Proto Compare
forall msg. Message msg => msg
defMessage
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareResult)
  (Proto Compare'CompareResult)
#result ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareResult)
  (Proto Compare'CompareResult)
-> Proto Compare'CompareResult -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Compare'CompareResult -> Proto Compare'CompareResult
forall msg. msg -> Proto msg
Proto Compare'CompareResult
Compare'GREATER
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareTarget)
  (Proto Compare'CompareTarget)
#target ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareTarget)
  (Proto Compare'CompareTarget)
-> Proto Compare'CompareTarget -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Compare'CompareTarget -> Proto Compare'CompareTarget
forall msg. msg -> Proto msg
Proto Compare'CompareTarget
Compare'VERSION
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter (Proto Compare) (Proto Compare) ByteString ByteString
#key ASetter (Proto Compare) (Proto Compare) ByteString ByteString
-> ByteString -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
versionKey
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter (Proto Compare) (Proto Compare) Int64 Int64
#version ASetter (Proto Compare) (Proto Compare) Int64 Int64
-> Int64 -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Int64
0

  getVersion :: Proto RequestOp
getVersion =
    Proto RequestOp
forall msg. Message msg => msg
defMessage Proto RequestOp
-> (Proto RequestOp -> Proto RequestOp) -> Proto RequestOp
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto RangeRequest)
  (Proto RangeRequest)
#requestRange ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto RangeRequest)
  (Proto RangeRequest)
-> Proto RangeRequest -> Proto RequestOp -> Proto RequestOp
forall s t a b. ASetter s t a b -> b -> s -> t
.~ (Proto RangeRequest
forall msg. Message msg => msg
defMessage Proto RangeRequest
-> (Proto RangeRequest -> Proto RangeRequest) -> Proto RangeRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
#key ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
-> ByteString -> Proto RangeRequest -> Proto RangeRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
versionKey)

  putVersion :: Proto RequestOp
putVersion =
    Proto RequestOp
forall msg. Message msg => msg
defMessage
      Proto RequestOp
-> (Proto RequestOp -> Proto RequestOp) -> Proto RequestOp
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto PutRequest)
  (Proto PutRequest)
#requestPut
        ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto PutRequest)
  (Proto PutRequest)
-> Proto PutRequest -> Proto RequestOp -> Proto RequestOp
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ( Proto PutRequest
forall msg. Message msg => msg
defMessage
              Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#key ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
versionKey
              Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#value ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ProtocolVersion -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' ProtocolVersion
ourVersion
           )

-- | Broadcast messages from a queue to the etcd cluster.
--
-- Retries on failure to 'putMessage' in case we are on a minority cluster or
-- when the grpc call timeouts.
--
-- Idempotent under transient 'GrpcDeadlineExceeded': 'putMessage' uses an
-- etcd transaction conditioned on the key's current 'mod_revision' matching
-- the last revision we successfully wrote. If a deadline-exceeded retry
-- arrives at etcd after the original request already committed, the compare
-- fails server-side (the mod_revision has advanced), the failure branch's
-- range tells us what the new revision is, and we move on without writing a
-- second time. So a retry whose original committed creates zero extra etcd
-- revisions and the watcher on each peer sees exactly one event per logical
-- broadcast. Same key namespace as master ('msg-\<host\>'), no disk growth.
broadcastMessages ::
  Tracer IO EtcdLog ->
  NetworkConfiguration ->
  -- | Used to identify sender.
  Host ->
  PersistentQueue IO msg ->
  IO ()
broadcastMessages :: forall msg.
Tracer IO EtcdLog
-> NetworkConfiguration -> Host -> PersistentQueue IO msg -> IO ()
broadcastMessages Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
ourHost PersistentQueue IO msg
queue = do
  -- Seed 'lastModRev' from etcd. With this in place the in-memory value
  -- always matches the server's view of our key when the loop starts:
  --
  --   * fresh process + fresh etcd → key absent → @lastModRev = 0@;
  --     'putMessage' compares against 0 (which etcd treats as "key does
  --     not exist") and the success branch creates the key.
  --   * fresh process + persisted etcd (e.g. Carol restart) → key
  --     present with some non-zero @mod_revision@ → @lastModRev@ starts
  --     at that revision; 'putMessage' compares against it and the
  --     success branch advances.
  --
  -- The init query removes the ambiguity that the older code had on
  -- @lastModRev == 0 + compare-fail@: with seeding, any compare-fail is
  -- unambiguously this peer's own deadline-exceeded retry, so the
  -- failure branch just adopts the new baseline and pops.
  Int64
initialModRev <- IO Int64
retryInitQuery
  TVar Int64
lastModRevVar <- String -> Int64 -> IO (TVar IO Int64)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"etcd-broadcast-last-mod-rev" Int64
initialModRev
  TVar (Maybe [(msg, ByteString)])
inFlightVar <- String
-> Maybe [(msg, ByteString)]
-> IO (TVar IO (Maybe [(msg, ByteString)]))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"etcd-broadcast-in-flight" Maybe [(msg, ByteString)]
forall a. Maybe a
Nothing
  Text -> IO () -> IO ()
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"broadcastMessages" (IO () -> IO ()) -> (IO () -> IO ()) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    -- Block for work before opening a connection, then keep it for
    -- 'maxPutsPerConnection' puts, waiting on it while the queue is dry. A
    -- handshake per burst costs ~0.5ms on the hot path and churns ephemeral
    -- ports. Messages are only popped after a successful put, so a recycled or
    -- failed connection never loses one.
    IO msg -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO msg -> IO ()) -> IO msg -> IO ()
forall a b. (a -> b) -> a -> b
$ PersistentQueue IO msg -> IO msg
forall (m :: * -> *) a. MonadSTM m => PersistentQueue m a -> m a
peekPersistentQueue PersistentQueue IO msg
queue
    (SomeException -> Maybe Text) -> IO () -> (Text -> IO ()) -> IO ()
forall e b a.
Exception e =>
(e -> Maybe b) -> IO a -> (b -> IO a) -> IO a
forall (m :: * -> *) e b a.
(MonadCatch m, Exception e) =>
(e -> Maybe b) -> m a -> (b -> m a) -> m a
catchJust SomeException -> Maybe Text
retryableEtcdError (TVar Int64 -> TVar (Maybe [(msg, ByteString)]) -> IO ()
sendPending TVar Int64
lastModRevVar TVar (Maybe [(msg, ByteString)])
inFlightVar) ((Text -> IO ()) -> IO ()) -> (Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Text
reason -> do
      Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer (EtcdLog -> IO ()) -> EtcdLog -> IO ()
forall a b. (a -> b) -> a -> b
$ BroadcastFailed{Text
$sel:reason:EtcdLog :: Text
reason :: Text
reason}
      DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1
 where
  sendPending :: TVar Int64 -> TVar (Maybe [(msg, ByteString)]) -> IO ()
sendPending TVar Int64
lastModRevVar TVar (Maybe [(msg, ByteString)])
inFlightVar =
    ConnParams -> Server -> (Connection -> IO ()) -> IO ()
forall a. ConnParams -> Server -> (Connection -> IO a) -> IO a
withConnection (Tracer IO EtcdLog -> Maybe Timeout -> ConnParams
connParams Tracer IO EtcdLog
tracer (Timeout -> Maybe Timeout
forall a. a -> Maybe a
Just (Timeout -> Maybe Timeout)
-> (TimeoutValue -> Timeout) -> TimeoutValue -> Maybe Timeout
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TimeoutUnit -> TimeoutValue -> Timeout
Timeout TimeoutUnit
Second (TimeoutValue -> Maybe Timeout) -> TimeoutValue -> Maybe Timeout
forall a b. (a -> b) -> a -> b
$ Word -> TimeoutValue
TimeoutValue Word
3)) (NetworkConfiguration -> Server
grpcServer NetworkConfiguration
config) ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
      let go :: Int -> IO ()
go Int
n
            | Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
0 = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            | Bool
otherwise = do
                -- Waiting here, rather than returning on an empty queue, is what
                -- makes a connection last 'maxPutsPerConnection' puts and not a
                -- single burst.
                IO msg -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO msg -> IO ()) -> IO msg -> IO ()
forall a b. (a -> b) -> a -> b
$ PersistentQueue IO msg -> IO msg
forall (m :: * -> *) a. MonadSTM m => PersistentQueue m a -> m a
peekPersistentQueue PersistentQueue IO msg
queue
                TVar IO (Maybe [(msg, ByteString)])
-> PersistentQueue IO msg
-> Int
-> Int
-> IO (Maybe [(msg, ByteString)])
forall (m :: * -> *) a.
MonadSTM m =>
TVar m (Maybe [(a, ByteString)])
-> PersistentQueue m a -> Int -> Int -> m (Maybe [(a, ByteString)])
nextPendingBatch TVar (Maybe [(msg, ByteString)])
TVar IO (Maybe [(msg, ByteString)])
inFlightVar PersistentQueue IO msg
queue Int
maxBatchCount Int
maxBatchBytes IO (Maybe [(msg, ByteString)])
-> (Maybe [(msg, ByteString)] -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                  -- Unreachable: we are the only consumer and just peeked.
                  Maybe [(msg, ByteString)]
Nothing -> Int -> IO ()
go Int
n
                  Just [(msg, ByteString)]
batch -> do
                    Tracer IO EtcdLog
-> Connection -> Host -> TVar IO Int64 -> ByteString -> IO ()
putMessage Tracer IO EtcdLog
tracer Connection
conn Host
ourHost TVar Int64
TVar IO Int64
lastModRevVar ([ByteString] -> ByteString
batchValue ([ByteString] -> ByteString) -> [ByteString] -> ByteString
forall a b. (a -> b) -> a -> b
$ (msg, ByteString) -> ByteString
forall a b. (a, b) -> b
snd ((msg, ByteString) -> ByteString)
-> [(msg, ByteString)] -> [ByteString]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [(msg, ByteString)]
batch)
                    Tracer IO EtcdLog
-> PersistentQueue IO msg -> [(msg, ByteString)] -> IO ()
forall (m :: * -> *) a.
(MonadSTM m, MonadIO m) =>
Tracer IO EtcdLog
-> PersistentQueue m a -> [(a, ByteString)] -> m ()
popBatchPersistentQueue Tracer IO EtcdLog
tracer PersistentQueue IO msg
queue [(msg, ByteString)]
batch
                    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO (Maybe [(msg, ByteString)])
-> Maybe [(msg, ByteString)] -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar (Maybe [(msg, ByteString)])
TVar IO (Maybe [(msg, ByteString)])
inFlightVar Maybe [(msg, ByteString)]
forall a. Maybe a
Nothing
                    Int -> IO ()
go (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1)
       in Int -> IO ()
go Int
maxPutsPerConnection

  -- Hedge against a connection wedging without raising. Not needed for
  -- https://github.com/cardano-scaling/hydra/issues/2167: that was seen while
  -- broadcast shared the connection with 'waitMessages'' watch stream, and a
  -- dedicated one sustains 20k puts.
  maxPutsPerConnection :: Int
maxPutsPerConnection = Int
1000 :: Int

  maxBatchCount :: Int
maxBatchCount = Int
50

  -- Keeps the value comfortably below etcd's default 1.5MiB request limit.
  maxBatchBytes :: Int
maxBatchBytes = Int
256 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1024
  -- Same retry shape as the broadcast loop, so we survive an etcd cluster that
  -- is still electing or that we briefly cannot reach.
  retryInitQuery :: IO Int64
retryInitQuery =
    (SomeException -> Maybe Text)
-> IO Int64 -> (Text -> IO Int64) -> IO Int64
forall e b a.
Exception e =>
(e -> Maybe b) -> IO a -> (b -> IO a) -> IO a
forall (m :: * -> *) e b a.
(MonadCatch m, Exception e) =>
(e -> Maybe b) -> m a -> (b -> m a) -> m a
catchJust SomeException -> Maybe Text
retryableEtcdError (Tracer IO EtcdLog -> NetworkConfiguration -> Host -> IO Int64
queryInitialModRev Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
ourHost) ((Text -> IO Int64) -> IO Int64) -> (Text -> IO Int64) -> IO Int64
forall a b. (a -> b) -> a -> b
$ \Text
reason -> do
      Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer (EtcdLog -> IO ()) -> EtcdLog -> IO ()
forall a b. (a -> b) -> a -> b
$ BroadcastFailed{Text
$sel:reason:EtcdLog :: Text
reason :: Text
reason}
      DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1
      IO Int64
retryInitQuery

-- | Query etcd for the current 'mod_revision' of this peer's broadcast key.
-- Returns 0 if the key does not yet exist. Used by 'broadcastMessages' to
-- seed its in-memory baseline so subsequent @compare mod_revision@ checks
-- match etcd's reality from the very first 'putMessage'.
queryInitialModRev ::
  Tracer IO EtcdLog ->
  NetworkConfiguration ->
  Host ->
  IO Int64
queryInitialModRev :: Tracer IO EtcdLog -> NetworkConfiguration -> Host -> IO Int64
queryInitialModRev Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
ourHost =
  ConnParams -> Server -> (Connection -> IO Int64) -> IO Int64
forall a. ConnParams -> Server -> (Connection -> IO a) -> IO a
withConnection (Tracer IO EtcdLog -> Maybe Timeout -> ConnParams
connParams Tracer IO EtcdLog
tracer (Timeout -> Maybe Timeout
forall a. a -> Maybe a
Just (Timeout -> Maybe Timeout)
-> (TimeoutValue -> Timeout) -> TimeoutValue -> Maybe Timeout
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TimeoutUnit -> TimeoutValue -> Timeout
Timeout TimeoutUnit
Second (TimeoutValue -> Maybe Timeout) -> TimeoutValue -> Maybe Timeout
forall a b. (a -> b) -> a -> b
$ Word -> TimeoutValue
TimeoutValue Word
3)) (NetworkConfiguration -> Server
grpcServer NetworkConfiguration
config) ((Connection -> IO Int64) -> IO Int64)
-> (Connection -> IO Int64) -> IO Int64
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
    Proto RangeResponse
res <- Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf KV "range")
-> Input (Protobuf KV "range")
-> IO (Output (Protobuf KV "range"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf KV "range")) Input (Protobuf KV "range")
Proto RangeRequest
req
    Int64 -> IO Int64
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int64 -> IO Int64) -> Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ Int64 -> Maybe Int64 -> Int64
forall a. a -> Maybe a -> a
fromMaybe Int64
0 (Proto RangeResponse
res Proto RangeResponse
-> Getting (First Int64) (Proto RangeResponse) Int64 -> Maybe Int64
forall s a. s -> Getting (First a) s a -> Maybe a
^? ([Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> Proto RangeResponse -> Const (First Int64) (Proto RangeResponse)
#kvs (([Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
 -> Proto RangeResponse
 -> Const (First Int64) (Proto RangeResponse))
-> ((Int64 -> Const (First Int64) Int64)
    -> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> Getting (First Int64) (Proto RangeResponse) Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto KeyValue -> Const (First Int64) (Proto KeyValue))
-> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse ((Proto KeyValue -> Const (First Int64) (Proto KeyValue))
 -> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> ((Int64 -> Const (First Int64) Int64)
    -> Proto KeyValue -> Const (First Int64) (Proto KeyValue))
-> (Int64 -> Const (First Int64) Int64)
-> [Proto KeyValue]
-> Const (First Int64) [Proto KeyValue]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> Const (First Int64) Int64)
-> Proto KeyValue -> Const (First Int64) (Proto KeyValue)
#modRevision)
 where
  key :: ByteString
key = forall a b. ConvertUtf8 a b => a -> b
encodeUtf8 @Text (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text
"msg-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Host -> Text
forall b a. (Show a, IsString b) => a -> b
show Host
ourHost
  req :: Proto RangeRequest
req = Proto RangeRequest
forall msg. Message msg => msg
defMessage Proto RangeRequest
-> (Proto RangeRequest -> Proto RangeRequest) -> Proto RangeRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
#key ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
-> ByteString -> Proto RangeRequest -> Proto RangeRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
key

-- | Broadcast a message to the etcd cluster.
--
-- Wraps the etcd @put@ in a @Txn@:
--
--   * compare: @mod_revision(key) == lastModRev@
--   * success: @put(key, value)@
--   * failure: @range(key)@ (so we can learn the actual @mod_revision@ if our
--     compare failed, e.g. because a previous attempt of ours committed
--     server-side after returning 'GrpcDeadlineExceeded' to the client)
--
-- 'lastModRevVar' is updated in both branches: from the response header on
-- a successful put, from the range result on a compare failure. Retries of
-- the same @msg@ after a deadline-exceeded converge to a single effective
-- revision rather than producing duplicate deliveries. Because
-- 'broadcastMessages' seeds 'lastModRev' from etcd at startup
-- ('queryInitialModRev'), a compare failure unambiguously means "this
-- peer's earlier attempt already wrote a later revision than we have
-- recorded" — i.e. the message is already delivered and the caller can pop.
--
-- __Cluster-reset behaviour (intentionally fatal):__ if the compare fails
-- and the @range@ branch returns no kvs, the etcd cluster has lost the
-- key we wrote against — either the data dir was wiped or the cluster
-- was replaced underneath us. We do not silently reseed: this is a
-- node-level event that should surface, not be papered over. 'putMessage'
-- calls 'fail', which kills the broadcast loop, propagates up to take
-- down the node, and on restart 'queryInitialModRev' re-seeds
-- 'lastModRev' from whatever state etcd actually has.
putMessage ::
  Tracer IO EtcdLog ->
  -- | Connection provided (and recycled) by 'broadcastMessages'.
  Connection ->
  -- | Used to identify sender.
  Host ->
  -- | The peer's last observed 'mod_revision' on its own broadcast key.
  TVar IO Int64 ->
  -- | Value to write, a batch of serialized messages (see 'batchValue').
  ByteString ->
  IO ()
putMessage :: Tracer IO EtcdLog
-> Connection -> Host -> TVar IO Int64 -> ByteString -> IO ()
putMessage Tracer IO EtcdLog
tracer Connection
conn Host
ourHost TVar IO Int64
lastModRevVar ByteString
value = do
  Int64
lastModRev <- TVar IO Int64 -> IO Int64
forall a. TVar IO a -> IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar IO Int64
lastModRevVar
  Proto TxnResponse
res <- Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf KV "txn")
-> Input (Protobuf KV "txn")
-> IO (Output (Protobuf KV "txn"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf KV "txn")) (Int64 -> Proto TxnRequest
txnReq Int64
lastModRev)
  if Proto TxnResponse
res Proto TxnResponse -> Getting Bool (Proto TxnResponse) Bool -> Bool
forall s a. s -> Getting a s a -> a
^. Getting Bool (Proto TxnResponse) Bool
#succeeded
    then
      -- Our compare matched and the put ran. The new mod_revision on
      -- our key equals the cluster revision returned in the response
      -- header.
      STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO Int64 -> Int64 -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar IO Int64
lastModRevVar (Proto TxnResponse
res Proto TxnResponse
-> Getting Int64 (Proto TxnResponse) Int64 -> Int64
forall s a. s -> Getting a s a -> a
^. (Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
-> Proto TxnResponse -> Const Int64 (Proto TxnResponse)
#header ((Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
 -> Proto TxnResponse -> Const Int64 (Proto TxnResponse))
-> ((Int64 -> Const Int64 Int64)
    -> Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
-> Getting Int64 (Proto TxnResponse) Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> Const Int64 Int64)
-> Proto ResponseHeader -> Const Int64 (Proto ResponseHeader)
#revision)
    else case Proto TxnResponse
res Proto TxnResponse
-> Getting (First Int64) (Proto TxnResponse) Int64 -> Maybe Int64
forall s a. s -> Getting (First a) s a -> Maybe a
^? ([Proto ResponseOp] -> Const (First Int64) [Proto ResponseOp])
-> Proto TxnResponse -> Const (First Int64) (Proto TxnResponse)
#responses (([Proto ResponseOp] -> Const (First Int64) [Proto ResponseOp])
 -> Proto TxnResponse -> Const (First Int64) (Proto TxnResponse))
-> ((Int64 -> Const (First Int64) Int64)
    -> [Proto ResponseOp] -> Const (First Int64) [Proto ResponseOp])
-> Getting (First Int64) (Proto TxnResponse) Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto ResponseOp -> Const (First Int64) (Proto ResponseOp))
-> [Proto ResponseOp] -> Const (First Int64) [Proto ResponseOp]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse ((Proto ResponseOp -> Const (First Int64) (Proto ResponseOp))
 -> [Proto ResponseOp] -> Const (First Int64) [Proto ResponseOp])
-> ((Int64 -> Const (First Int64) Int64)
    -> Proto ResponseOp -> Const (First Int64) (Proto ResponseOp))
-> (Int64 -> Const (First Int64) Int64)
-> [Proto ResponseOp]
-> Const (First Int64) [Proto ResponseOp]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto RangeResponse -> Const (First Int64) (Proto RangeResponse))
-> Proto ResponseOp -> Const (First Int64) (Proto ResponseOp)
#responseRange ((Proto RangeResponse -> Const (First Int64) (Proto RangeResponse))
 -> Proto ResponseOp -> Const (First Int64) (Proto ResponseOp))
-> Getting (First Int64) (Proto RangeResponse) Int64
-> (Int64 -> Const (First Int64) Int64)
-> Proto ResponseOp
-> Const (First Int64) (Proto ResponseOp)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ([Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> Proto RangeResponse -> Const (First Int64) (Proto RangeResponse)
#kvs (([Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
 -> Proto RangeResponse
 -> Const (First Int64) (Proto RangeResponse))
-> ((Int64 -> Const (First Int64) Int64)
    -> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> Getting (First Int64) (Proto RangeResponse) Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto KeyValue -> Const (First Int64) (Proto KeyValue))
-> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse ((Proto KeyValue -> Const (First Int64) (Proto KeyValue))
 -> [Proto KeyValue] -> Const (First Int64) [Proto KeyValue])
-> ((Int64 -> Const (First Int64) Int64)
    -> Proto KeyValue -> Const (First Int64) (Proto KeyValue))
-> (Int64 -> Const (First Int64) Int64)
-> [Proto KeyValue]
-> Const (First Int64) [Proto KeyValue]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> Const (First Int64) Int64)
-> Proto KeyValue -> Const (First Int64) (Proto KeyValue)
#modRevision of
      Just Int64
observedModRev -> do
        -- Compare failed. Since 'broadcastMessages' seeded 'lastModRev'
        -- from etcd at startup and we are the only writer to our key,
        -- the only way 'mod_revision' moved past 'lastModRev' is an
        -- earlier attempt of ours committing server-side despite a
        -- 'GrpcDeadlineExceeded' to the client. The message has been
        -- delivered; adopt the new baseline and let the outer loop pop.
        Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer BroadcastDeduped{$sel:previousModRev:EtcdLog :: Int64
previousModRev = Int64
lastModRev, Int64
observedModRev :: Int64
$sel:observedModRev:EtcdLog :: Int64
observedModRev}
        STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO Int64 -> Int64 -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar IO Int64
lastModRevVar Int64
observedModRev
      Maybe Int64
Nothing ->
        -- Compare failed AND range came back empty: etcd has no record
        -- of our key. Unreachable in normal operation (only we write to
        -- our key; nothing deletes it). If we ever do hit this, the
        -- safe move is to crash loudly — the surrounding race kills the
        -- node and a fresh start re-runs 'queryInitialModRev' against
        -- whatever state etcd actually has.
        String -> IO ()
forall a. String -> IO a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$
          String
"putMessage: compare against mod_revision "
            String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int64 -> String
forall b a. (Show a, IsString b) => a -> b
show Int64
lastModRev
            String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" failed but our broadcast key has no current value in etcd"
 where
  key :: ByteString
key = forall a b. ConvertUtf8 a b => a -> b
encodeUtf8 @Text (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text
"msg-" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Host -> Text
forall b a. (Show a, IsString b) => a -> b
show Host
ourHost

  -- Compare: mod_revision(key) == lastModRev
  modRevMatches :: Int64 -> Proto Compare
modRevMatches Int64
lastModRev =
    Proto Compare
forall msg. Message msg => msg
defMessage
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareResult)
  (Proto Compare'CompareResult)
#result ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareResult)
  (Proto Compare'CompareResult)
-> Proto Compare'CompareResult -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Compare'CompareResult -> Proto Compare'CompareResult
forall msg. msg -> Proto msg
Proto Compare'CompareResult
Compare'EQUAL
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareTarget)
  (Proto Compare'CompareTarget)
#target ASetter
  (Proto Compare)
  (Proto Compare)
  (Proto Compare'CompareTarget)
  (Proto Compare'CompareTarget)
-> Proto Compare'CompareTarget -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Compare'CompareTarget -> Proto Compare'CompareTarget
forall msg. msg -> Proto msg
Proto Compare'CompareTarget
Compare'MOD
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter (Proto Compare) (Proto Compare) ByteString ByteString
#key ASetter (Proto Compare) (Proto Compare) ByteString ByteString
-> ByteString -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
key
      Proto Compare -> (Proto Compare -> Proto Compare) -> Proto Compare
forall a b. a -> (a -> b) -> b
& ASetter (Proto Compare) (Proto Compare) Int64 Int64
#modRevision ASetter (Proto Compare) (Proto Compare) Int64 Int64
-> Int64 -> Proto Compare -> Proto Compare
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Int64
lastModRev

  putReqOp :: Proto RequestOp
putReqOp =
    Proto RequestOp
forall msg. Message msg => msg
defMessage
      Proto RequestOp
-> (Proto RequestOp -> Proto RequestOp) -> Proto RequestOp
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto PutRequest)
  (Proto PutRequest)
#requestPut
        ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto PutRequest)
  (Proto PutRequest)
-> Proto PutRequest -> Proto RequestOp -> Proto RequestOp
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ( Proto PutRequest
forall msg. Message msg => msg
defMessage
              Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#key ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
key
              Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#value ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
value
           )

  rangeReqOp :: Proto RequestOp
rangeReqOp =
    Proto RequestOp
forall msg. Message msg => msg
defMessage Proto RequestOp
-> (Proto RequestOp -> Proto RequestOp) -> Proto RequestOp
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto RangeRequest)
  (Proto RangeRequest)
#requestRange ASetter
  (Proto RequestOp)
  (Proto RequestOp)
  (Proto RangeRequest)
  (Proto RangeRequest)
-> Proto RangeRequest -> Proto RequestOp -> Proto RequestOp
forall s t a b. ASetter s t a b -> b -> s -> t
.~ (Proto RangeRequest
forall msg. Message msg => msg
defMessage Proto RangeRequest
-> (Proto RangeRequest -> Proto RangeRequest) -> Proto RangeRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
#key ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
-> ByteString -> Proto RangeRequest -> Proto RangeRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
key)

  txnReq :: Int64 -> Proto TxnRequest
txnReq Int64
lastModRev =
    Proto TxnRequest
forall msg. Message msg => msg
defMessage
      Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto Compare]
  [Proto Compare]
#compare ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto Compare]
  [Proto Compare]
-> [Proto Compare] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Int64 -> Proto Compare
modRevMatches Int64
lastModRev]
      Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
#success ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
-> [Proto RequestOp] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Proto RequestOp
putReqOp]
      Proto TxnRequest
-> (Proto TxnRequest -> Proto TxnRequest) -> Proto TxnRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
#failure ASetter
  (Proto TxnRequest)
  (Proto TxnRequest)
  [Proto RequestOp]
  [Proto RequestOp]
-> [Proto RequestOp] -> Proto TxnRequest -> Proto TxnRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ [Proto RequestOp
rangeReqOp]

-- | Assemble the etcd value for a batch of already-serialized messages: a
-- CBOR list reusing each message's encoding verbatim. Uses the same
-- indefinite-length list format as cardano-binary's list instances, so the
-- value decodes as @[msg]@ on the receiving side.
batchValue :: [ByteString] -> ByteString
batchValue :: [ByteString] -> ByteString
batchValue [ByteString]
encodedItems =
  Encoding -> ByteString
CBOR.toStrictByteString (Encoding -> ByteString) -> Encoding -> ByteString
forall a b. (a -> b) -> a -> b
$
    Encoding
CBOR.encodeListLenIndef
      Encoding -> Encoding -> Encoding
forall a. Semigroup a => a -> a -> a
<> (ByteString -> Encoding) -> [ByteString] -> Encoding
forall m a. Monoid m => (a -> m) -> [a] -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap ByteString -> Encoding
CBOR.encodePreEncoded [ByteString]
encodedItems
      Encoding -> Encoding -> Encoding
forall a. Semigroup a => a -> a -> a
<> Encoding
CBOR.encodeBreak

-- | Fetch and wait for messages from the etcd cluster.
waitMessages ::
  forall msg.
  FromCBOR msg =>
  Tracer IO EtcdLog ->
  Connection ->
  FilePath ->
  NetworkCallback msg IO ->
  IO ()
waitMessages :: forall msg.
FromCBOR msg =>
Tracer IO EtcdLog
-> Connection -> String -> NetworkCallback msg IO -> IO ()
waitMessages Tracer IO EtcdLog
tracer Connection
conn String
directory NetworkCallback{msg -> IO ()
deliver :: msg -> IO ()
$sel:deliver:NetworkCallback :: forall msg (m :: * -> *). NetworkCallback msg m -> msg -> m ()
deliver} =
  Text -> IO () -> IO ()
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"waitMessages" (IO () -> IO ()) -> (IO () -> IO ()) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    -- Restart a failed watch from the last known revision, same as one that
    -- ended cleanly. Without this any connection blip kills the node.
    (SomeException -> Maybe Text) -> IO () -> (Text -> IO ()) -> IO ()
forall e b a.
Exception e =>
(e -> Maybe b) -> IO a -> (b -> IO a) -> IO a
forall (m :: * -> *) e b a.
(MonadCatch m, Exception e) =>
(e -> Maybe b) -> m a -> (b -> m a) -> m a
catchJust SomeException -> Maybe Text
retryableEtcdError IO ()
watch ((Text -> IO ()) -> IO ()) -> (Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Text
reason ->
      Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer WatchFailed{Text
$sel:reason:EtcdLog :: Text
reason :: Text
reason}
    -- Wait before re-trying
    DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1
 where
  -- NOTE: We have not observed the watch (subscription) fail even when peers
  -- leave and we end up on a minority cluster.
  watch :: IO ()
watch =
    Connection
-> ClientHandler'
     'BiDiStreaming (ReaderT Connection IO) (Protobuf Watch "watch")
-> ((NextElem (Input (Protobuf Watch "watch")) -> IO ())
    -> IO (NextElem (Output (Protobuf Watch "watch"))) -> IO ())
-> IO ()
forall {k} (rpc :: k) (m :: * -> *) r.
MonadIO m =>
Connection
-> ClientHandler' 'BiDiStreaming (ReaderT Connection m) rpc
-> ((NextElem (Input rpc) -> m ())
    -> m (NextElem (Output rpc)) -> m r)
-> m r
biDiStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf Watch "watch")) (((NextElem (Input (Protobuf Watch "watch")) -> IO ())
  -> IO (NextElem (Output (Protobuf Watch "watch"))) -> IO ())
 -> IO ())
-> ((NextElem (Input (Protobuf Watch "watch")) -> IO ())
    -> IO (NextElem (Output (Protobuf Watch "watch"))) -> IO ())
-> IO ()
forall a b. (a -> b) -> a -> b
$ \NextElem (Input (Protobuf Watch "watch")) -> IO ()
send IO (NextElem (Output (Protobuf Watch "watch")))
recv -> do
      Natural
revision <- String -> IO Natural
forall (m :: * -> *). MonadIO m => String -> m Natural
getLastKnownRevision String
directory
      let startRevision :: Int64
startRevision = Natural -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Natural
revision Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
1)
      Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer WatchMessagesStartRevision{Int64
startRevision :: Int64
$sel:startRevision:EtcdLog :: Int64
startRevision}
      -- NOTE: Request all keys starting with 'msg'. See also section KeyRanges
      -- in https://etcd.io/docs/v3.5/learning/api/#key-value-api
      let watchRequest :: Proto WatchCreateRequest
watchRequest =
            Proto WatchCreateRequest
forall msg. Message msg => msg
defMessage
              Proto WatchCreateRequest
-> (Proto WatchCreateRequest -> Proto WatchCreateRequest)
-> Proto WatchCreateRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
  ByteString
  ByteString
#key ASetter
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
  ByteString
  ByteString
-> ByteString
-> Proto WatchCreateRequest
-> Proto WatchCreateRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
"msg"
              Proto WatchCreateRequest
-> (Proto WatchCreateRequest -> Proto WatchCreateRequest)
-> Proto WatchCreateRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
  ByteString
  ByteString
#rangeEnd ASetter
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
  ByteString
  ByteString
-> ByteString
-> Proto WatchCreateRequest
-> Proto WatchCreateRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
"msh" -- NOTE: g+1 to query prefixes
              Proto WatchCreateRequest
-> (Proto WatchCreateRequest -> Proto WatchCreateRequest)
-> Proto WatchCreateRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto WatchCreateRequest) (Proto WatchCreateRequest) Int64 Int64
#startRevision ASetter
  (Proto WatchCreateRequest) (Proto WatchCreateRequest) Int64 Int64
-> Int64 -> Proto WatchCreateRequest -> Proto WatchCreateRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Natural -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Natural
revision Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
1)
      NextElem (Input (Protobuf Watch "watch")) -> IO ()
NextElem (Proto WatchRequest) -> IO ()
send (NextElem (Proto WatchRequest) -> IO ())
-> (Proto WatchRequest -> NextElem (Proto WatchRequest))
-> Proto WatchRequest
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Proto WatchRequest -> NextElem (Proto WatchRequest)
forall a. a -> NextElem a
NextElem (Proto WatchRequest -> IO ()) -> Proto WatchRequest -> IO ()
forall a b. (a -> b) -> a -> b
$ Proto WatchRequest
forall msg. Message msg => msg
defMessage Proto WatchRequest
-> (Proto WatchRequest -> Proto WatchRequest) -> Proto WatchRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto WatchRequest)
  (Proto WatchRequest)
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
#createRequest ASetter
  (Proto WatchRequest)
  (Proto WatchRequest)
  (Proto WatchCreateRequest)
  (Proto WatchCreateRequest)
-> Proto WatchCreateRequest
-> Proto WatchRequest
-> Proto WatchRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Proto WatchCreateRequest
watchRequest
      (NextElem (Proto WatchRequest) -> IO ())
-> IO (NextElem (Proto WatchResponse)) -> IO ()
loop NextElem (Input (Protobuf Watch "watch")) -> IO ()
NextElem (Proto WatchRequest) -> IO ()
send IO (NextElem (Output (Protobuf Watch "watch")))
IO (NextElem (Proto WatchResponse))
recv

  loop :: (NextElem (Proto WatchRequest) -> IO ())
-> IO (NextElem (Proto WatchResponse)) -> IO ()
loop NextElem (Proto WatchRequest) -> IO ()
send IO (NextElem (Proto WatchResponse))
recv =
    IO (NextElem (Proto WatchResponse))
recv IO (NextElem (Proto WatchResponse))
-> (NextElem (Proto WatchResponse) -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      NextElem (Proto WatchResponse)
NoNextElem -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      NextElem Proto WatchResponse
res ->
        if Proto WatchResponse
res Proto WatchResponse
-> Getting Bool (Proto WatchResponse) Bool -> Bool
forall s a. s -> Getting a s a -> a
^. Getting Bool (Proto WatchResponse) Bool
#canceled
          then do
            let compactRevision :: Int64
compactRevision = Proto WatchResponse
res Proto WatchResponse
-> Getting Int64 (Proto WatchResponse) Int64 -> Int64
forall s a. s -> Getting a s a -> a
^. Getting Int64 (Proto WatchResponse) Int64
#compactRevision
            Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer WatchMessagesFallbackTo{Int64
compactRevision :: Int64
$sel:compactRevision:EtcdLog :: Int64
compactRevision}
            String -> Natural -> IO ()
forall (m :: * -> *). MonadIO m => String -> Natural -> m ()
putLastKnownRevision String
directory (Natural -> IO ()) -> (Int64 -> Natural) -> Int64 -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int64 -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> IO ()) -> Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ (Int64
compactRevision Int64 -> Int64 -> Int64
forall a. Num a => a -> a -> a
- Int64
1) Int64 -> Int64 -> Int64
forall a. Ord a => a -> a -> a
`max` Int64
0
            -- Gracefully close watch stream
            NextElem (Proto WatchRequest) -> IO ()
send NextElem (Proto WatchRequest)
forall a. NextElem a
NoNextElem
          else do
            let revision :: Int64
revision = Proto WatchResponse
res Proto WatchResponse
-> Getting Int64 (Proto WatchResponse) Int64 -> Int64
forall s a. s -> Getting a s a -> a
^. (Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
-> Proto WatchResponse -> Const Int64 (Proto WatchResponse)
#header ((Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
 -> Proto WatchResponse -> Const Int64 (Proto WatchResponse))
-> ((Int64 -> Const Int64 Int64)
    -> Proto ResponseHeader -> Const Int64 (Proto ResponseHeader))
-> Getting Int64 (Proto WatchResponse) Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int64 -> Const Int64 Int64)
-> Proto ResponseHeader -> Const Int64 (Proto ResponseHeader)
#revision
            String -> Natural -> IO ()
forall (m :: * -> *). MonadIO m => String -> Natural -> m ()
putLastKnownRevision String
directory (Natural -> IO ()) -> (Int64 -> Natural) -> Int64 -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int64 -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> IO ()) -> Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ Int64
revision Int64 -> Int64 -> Int64
forall a. Ord a => a -> a -> a
`max` Int64
0
            [Proto Event] -> (Proto Event -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (Proto WatchResponse
res Proto WatchResponse
-> Getting [Proto Event] (Proto WatchResponse) [Proto Event]
-> [Proto Event]
forall s a. s -> Getting a s a -> a
^. Getting [Proto Event] (Proto WatchResponse) [Proto Event]
#events) Proto Event -> IO ()
process
            (NextElem (Proto WatchRequest) -> IO ())
-> IO (NextElem (Proto WatchResponse)) -> IO ()
loop NextElem (Proto WatchRequest) -> IO ()
send IO (NextElem (Proto WatchResponse))
recv

  process :: Proto Event -> IO ()
process Proto Event
event = do
    let value :: ByteString
value = Proto Event
event Proto Event
-> Getting ByteString (Proto Event) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. (Proto KeyValue -> Const ByteString (Proto KeyValue))
-> Proto Event -> Const ByteString (Proto Event)
#kv ((Proto KeyValue -> Const ByteString (Proto KeyValue))
 -> Proto Event -> Const ByteString (Proto Event))
-> Getting ByteString (Proto KeyValue) ByteString
-> Getting ByteString (Proto Event) ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Getting ByteString (Proto KeyValue) ByteString
#value
    -- Broadcast values carry a batch of messages per revision (see
    -- 'batchValue'). Watch catch-up can still replay single-message values
    -- written before an upgrade, so fall back to decoding one message.
    case forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' @[msg] ByteString
value of
      Right [msg]
msgs -> [msg] -> (msg -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [msg]
msgs msg -> IO ()
deliver
      Left DecoderError
err ->
        case ByteString -> Either DecoderError msg
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' ByteString
value of
          Right msg
msg -> msg -> IO ()
deliver msg
msg
          Left DecoderError
_ ->
            Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith
              Tracer IO EtcdLog
tracer
              FailedToDecodeValue
                { $sel:key:EtcdLog :: Text
key = ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 (ByteString -> Text) -> ByteString -> Text
forall a b. (a -> b) -> a -> b
$ Proto Event
event Proto Event
-> Getting ByteString (Proto Event) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. (Proto KeyValue -> Const ByteString (Proto KeyValue))
-> Proto Event -> Const ByteString (Proto Event)
#kv ((Proto KeyValue -> Const ByteString (Proto KeyValue))
 -> Proto Event -> Const ByteString (Proto Event))
-> Getting ByteString (Proto KeyValue) ByteString
-> Getting ByteString (Proto Event) ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Getting ByteString (Proto KeyValue) ByteString
#key
                , $sel:value:EtcdLog :: Text
value = ByteString -> Text
encodeBase16 ByteString
value
                , $sel:reason:EtcdLog :: Text
reason = DecoderError -> Text
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err
                }

-- | The persisted watch revision could not be read.
--
-- Fatal on purpose, and deliberately not a 'retryableEtcdError': re-reading the
-- same bad file cannot succeed, and carrying on from revision 0 would rewind the
-- watch to the beginning of history. Deleting the file is the fix, so say so.
data LastKnownRevisionException
  = -- | The file exists but does not hold a revision.
    InvalidLastKnownRevision FilePath
  | -- | The file exists but could not be read.
    UnreadableLastKnownRevision FilePath String
  deriving stock (LastKnownRevisionException -> LastKnownRevisionException -> Bool
(LastKnownRevisionException -> LastKnownRevisionException -> Bool)
-> (LastKnownRevisionException
    -> LastKnownRevisionException -> Bool)
-> Eq LastKnownRevisionException
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: LastKnownRevisionException -> LastKnownRevisionException -> Bool
== :: LastKnownRevisionException -> LastKnownRevisionException -> Bool
$c/= :: LastKnownRevisionException -> LastKnownRevisionException -> Bool
/= :: LastKnownRevisionException -> LastKnownRevisionException -> Bool
Eq, Int -> LastKnownRevisionException -> String -> String
[LastKnownRevisionException] -> String -> String
LastKnownRevisionException -> String
(Int -> LastKnownRevisionException -> String -> String)
-> (LastKnownRevisionException -> String)
-> ([LastKnownRevisionException] -> String -> String)
-> Show LastKnownRevisionException
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> LastKnownRevisionException -> String -> String
showsPrec :: Int -> LastKnownRevisionException -> String -> String
$cshow :: LastKnownRevisionException -> String
show :: LastKnownRevisionException -> String
$cshowList :: [LastKnownRevisionException] -> String -> String
showList :: [LastKnownRevisionException] -> String -> String
Show)

instance Exception LastKnownRevisionException where
  displayException :: LastKnownRevisionException -> String
displayException = \case
    InvalidLastKnownRevision String
file ->
      String
"Failed to load last known revision: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
file String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" does not hold a revision. " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String -> String
remediation String
file
    UnreadableLastKnownRevision String
file String
reason ->
      String
"Failed to load last known revision: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
reason String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
". " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String -> String
remediation String
file
   where
    remediation :: FilePath -> String
    remediation :: String -> String
remediation String
file =
      String
"Delete " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
file String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" to re-sync from the last compacted revision."

getLastKnownRevision :: MonadIO m => FilePath -> m Natural
getLastKnownRevision :: forall (m :: * -> *). MonadIO m => String -> m Natural
getLastKnownRevision String
directory = do
  IO Natural -> m Natural
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Natural -> m Natural) -> IO Natural -> m Natural
forall a b. (a -> b) -> a -> b
$
    IO (Maybe Natural) -> IO (Either IOError (Maybe Natural))
forall e a. Exception e => IO a -> IO (Either e a)
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try (String -> IO (Maybe Natural)
forall a. FromJSON a => String -> IO (Maybe a)
decodeFileStrict' String
file) IO (Either IOError (Maybe Natural))
-> (Either IOError (Maybe Natural) -> IO Natural) -> IO Natural
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      -- NOTE: A 'Nothing' here means the file exists but holds no revision.
      -- Silently treating that as 0 would restart the watch from the beginning
      -- of history, which etcd then cancels with a compactRevision, so fail
      -- loudly instead. 'putLastKnownRevision' writes atomically, so a killed
      -- process can no longer produce this.
      Right Maybe Natural
Nothing -> LastKnownRevisionException -> IO Natural
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (LastKnownRevisionException -> IO Natural)
-> LastKnownRevisionException -> IO Natural
forall a b. (a -> b) -> a -> b
$ String -> LastKnownRevisionException
InvalidLastKnownRevision String
file
      Right (Just Natural
rev) -> Natural -> IO Natural
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
rev
      Left (IOError
e :: IOException)
        | IOError -> Bool
isDoesNotExistError IOError
e -> Natural -> IO Natural
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
0
        | Bool
otherwise -> LastKnownRevisionException -> IO Natural
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (LastKnownRevisionException -> IO Natural)
-> (String -> LastKnownRevisionException) -> String -> IO Natural
forall b c a. (b -> c) -> (a -> b) -> a -> c
. String -> String -> LastKnownRevisionException
UnreadableLastKnownRevision String
file (String -> IO Natural) -> String -> IO Natural
forall a b. (a -> b) -> a -> b
$ IOError -> String
forall e. Exception e => e -> String
displayException IOError
e
 where
  file :: String
file = String
directory String -> String -> String
</> String
"last-known-revision"

-- | Record the revision, atomically.
--
-- NOTE: Written to a temporary file and renamed into place, because a plain
-- 'encodeFile' is not atomic: killing the node mid-write (which happens
-- routinely, both in tests and on restart) would otherwise leave a truncated
-- file that 'getLastKnownRevision' cannot parse.
--
-- NOTE: The rename is not fsynced, so an OS crash can still land it ahead of the
-- data blocks. Closing that window costs an fsync per call, and this runs once
-- per watch response, i.e. per received broadcast. The resulting file is instead
-- handled on the next start, by 'LastKnownRevisionException' and its hint.
putLastKnownRevision :: MonadIO m => FilePath -> Natural -> m ()
putLastKnownRevision :: forall (m :: * -> *). MonadIO m => String -> Natural -> m ()
putLastKnownRevision String
directory Natural
rev = do
  IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    String -> Natural -> IO ()
forall a. ToJSON a => String -> a -> IO ()
encodeFile String
tmpFile Natural
rev
    String -> String -> IO ()
renameFile String
tmpFile String
file
 where
  file :: String
file = String
directory String -> String -> String
</> String
"last-known-revision"
  tmpFile :: String
tmpFile = String
file String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
".tmp"

-- | Write a well-known key to indicate being alive, keep it alive using a lease
-- and poll other peers entries to yield connectivity events. While doing so,
-- overall network connectivity is determined from the ability to read/write to
-- the cluster.
pollConnectivity ::
  Tracer IO EtcdLog ->
  Connection ->
  -- | Local host
  Host ->
  NetworkCallback msg IO ->
  IO ()
pollConnectivity :: forall msg.
Tracer IO EtcdLog
-> Connection -> Host -> NetworkCallback msg IO -> IO ()
pollConnectivity Tracer IO EtcdLog
tracer Connection
conn Host
advertise NetworkCallback{Connectivity -> IO ()
$sel:onConnectivity:NetworkCallback :: forall msg (m :: * -> *).
NetworkCallback msg m -> Connectivity -> m ()
onConnectivity :: Connectivity -> IO ()
onConnectivity} = do
  TVar [Host]
seenAliveVar <- String -> [Host] -> IO (TVar IO [Host])
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"etcd-seen-alive" []
  Text -> IO () -> IO ()
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"pollConnectivity" (IO () -> IO ()) -> (IO () -> IO ()) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    (SomeException -> Maybe Text) -> IO () -> (Text -> IO ()) -> IO ()
forall e b a.
Exception e =>
(e -> Maybe b) -> IO a -> (b -> IO a) -> IO a
forall (m :: * -> *) e b a.
(MonadCatch m, Exception e) =>
(e -> Maybe b) -> m a -> (b -> m a) -> m a
catchJust SomeException -> Maybe Text
retryableEtcdError (TVar [Host] -> IO ()
poll TVar [Host]
seenAliveVar) (TVar [Host] -> Text -> IO ()
onConnectionLost TVar [Host]
seenAliveVar)
 where
  poll :: TVar [Host] -> IO ()
poll TVar [Host]
seenAliveVar = do
    Int64
leaseId <- IO Int64
createLease
    -- If we can create a lease, we are connected
    Connectivity -> IO ()
onConnectivity Connectivity
NetworkConnected
    -- Write our alive key using lease
    Int64 -> IO ()
writeAlive Int64
leaseId
    Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer CreatedLease{Int64
leaseId :: Int64
$sel:leaseId:EtcdLog :: Int64
leaseId}
    Int64 -> (IO Int64 -> IO ()) -> IO ()
withKeepAlive Int64
leaseId (TVar [Host] -> IO Int64 -> IO ()
aliveLoop TVar [Host]
seenAliveVar)

  aliveLoop :: TVar [Host] -> IO Int64 -> IO ()
aliveLoop TVar [Host]
seenAliveVar IO Int64
keepAlive = do
    -- Keep our lease alive
    Int64
ttlRemaining <- IO Int64
keepAlive
    if Int64
ttlRemaining Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
<= Int64
0
      then
        -- The keep alive did not work as no time to live remaining. Get a new lease instead
        Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer LowLeaseTTL{Int64
ttlRemaining :: Int64
$sel:ttlRemaining:EtcdLog :: Int64
ttlRemaining}
      else do
        -- Determine alive peers
        [Host]
alive <- IO [Host]
getAlive
        let othersAlive :: [Host]
othersAlive = [Host]
alive [Host] -> [Host] -> [Host]
forall a. Eq a => [a] -> [a] -> [a]
\\ [Host
advertise]
        [Host]
seenAlive <- STM IO [Host] -> IO [Host]
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO [Host] -> IO [Host]) -> STM IO [Host] -> IO [Host]
forall a b. (a -> b) -> a -> b
$ TVar IO [Host] -> [Host] -> STM IO [Host]
forall a. TVar IO a -> a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m a
swapTVar TVar [Host]
TVar IO [Host]
seenAliveVar [Host]
othersAlive
        [Host] -> (Host -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ ([Host]
othersAlive [Host] -> [Host] -> [Host]
forall a. Eq a => [a] -> [a] -> [a]
\\ [Host]
seenAlive) ((Host -> IO ()) -> IO ()) -> (Host -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ Connectivity -> IO ()
onConnectivity (Connectivity -> IO ()) -> (Host -> Connectivity) -> Host -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Host -> Connectivity
PeerConnected
        [Host] -> (Host -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ ([Host]
seenAlive [Host] -> [Host] -> [Host]
forall a. Eq a => [a] -> [a] -> [a]
\\ [Host]
othersAlive) ((Host -> IO ()) -> IO ()) -> (Host -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ Connectivity -> IO ()
onConnectivity (Connectivity -> IO ()) -> (Host -> Connectivity) -> Host -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Host -> Connectivity
PeerDisconnected
        DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1
        TVar [Host] -> IO Int64 -> IO ()
aliveLoop TVar [Host]
seenAliveVar IO Int64
keepAlive

  -- Report the network as down and let the outer loop take a fresh lease.
  onConnectionLost :: TVar [Host] -> Text -> IO ()
onConnectionLost TVar [Host]
seenAliveVar Text
_reason = do
    Connectivity -> IO ()
onConnectivity Connectivity
NetworkDisconnected
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO [Host] -> [Host] -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar [Host]
TVar IO [Host]
seenAliveVar []
    DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
1

  createLease :: IO Int64
createLease = Text -> IO Int64 -> IO Int64
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"createLease" (IO Int64 -> IO Int64) -> IO Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ do
    Proto LeaseGrantResponse
leaseResponse <-
      Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf Lease "leaseGrant")
-> Input (Protobuf Lease "leaseGrant")
-> IO (Output (Protobuf Lease "leaseGrant"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf Lease "leaseGrant")) (Input (Protobuf Lease "leaseGrant")
 -> IO (Output (Protobuf Lease "leaseGrant")))
-> Input (Protobuf Lease "leaseGrant")
-> IO (Output (Protobuf Lease "leaseGrant"))
forall a b. (a -> b) -> a -> b
$
        Proto LeaseGrantRequest
forall msg. Message msg => msg
defMessage Proto LeaseGrantRequest
-> (Proto LeaseGrantRequest -> Proto LeaseGrantRequest)
-> Proto LeaseGrantRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto LeaseGrantRequest) (Proto LeaseGrantRequest) Int64 Int64
#ttl ASetter
  (Proto LeaseGrantRequest) (Proto LeaseGrantRequest) Int64 Int64
-> Int64 -> Proto LeaseGrantRequest -> Proto LeaseGrantRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Int64
3
    Int64 -> IO Int64
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int64 -> IO Int64) -> Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ Proto LeaseGrantResponse
leaseResponse Proto LeaseGrantResponse
-> Getting Int64 (Proto LeaseGrantResponse) Int64 -> Int64
forall s a. s -> Getting a s a -> a
^. Getting Int64 (Proto LeaseGrantResponse) Int64
#id

  withKeepAlive :: Int64 -> (IO Int64 -> IO ()) -> IO ()
withKeepAlive Int64
leaseId IO Int64 -> IO ()
action = do
    Connection
-> ClientHandler'
     'BiDiStreaming
     (ReaderT Connection IO)
     (Protobuf Lease "leaseKeepAlive")
-> ((NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ())
    -> IO (NextElem (Output (Protobuf Lease "leaseKeepAlive")))
    -> IO ())
-> IO ()
forall {k} (rpc :: k) (m :: * -> *) r.
MonadIO m =>
Connection
-> ClientHandler' 'BiDiStreaming (ReaderT Connection m) rpc
-> ((NextElem (Input rpc) -> m ())
    -> m (NextElem (Output rpc)) -> m r)
-> m r
biDiStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf Lease "leaseKeepAlive")) (((NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ())
  -> IO (NextElem (Output (Protobuf Lease "leaseKeepAlive")))
  -> IO ())
 -> IO ())
-> ((NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ())
    -> IO (NextElem (Output (Protobuf Lease "leaseKeepAlive")))
    -> IO ())
-> IO ()
forall a b. (a -> b) -> a -> b
$ \NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ()
send IO (NextElem (Output (Protobuf Lease "leaseKeepAlive")))
recv -> do
      IO () -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO () -> IO ()) -> (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IO Int64 -> IO ()
action (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ()
send (NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ())
-> NextElem (Input (Protobuf Lease "leaseKeepAlive")) -> IO ()
forall a b. (a -> b) -> a -> b
$ Input (Protobuf Lease "leaseKeepAlive")
-> NextElem (Input (Protobuf Lease "leaseKeepAlive"))
forall a. a -> NextElem a
NextElem (Input (Protobuf Lease "leaseKeepAlive")
 -> NextElem (Input (Protobuf Lease "leaseKeepAlive")))
-> Input (Protobuf Lease "leaseKeepAlive")
-> NextElem (Input (Protobuf Lease "leaseKeepAlive"))
forall a b. (a -> b) -> a -> b
$ Proto LeaseKeepAliveRequest
forall msg. Message msg => msg
defMessage Proto LeaseKeepAliveRequest
-> (Proto LeaseKeepAliveRequest -> Proto LeaseKeepAliveRequest)
-> Proto LeaseKeepAliveRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto LeaseKeepAliveRequest)
  (Proto LeaseKeepAliveRequest)
  Int64
  Int64
#id ASetter
  (Proto LeaseKeepAliveRequest)
  (Proto LeaseKeepAliveRequest)
  Int64
  Int64
-> Int64
-> Proto LeaseKeepAliveRequest
-> Proto LeaseKeepAliveRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Int64
leaseId
        IO (NextElem (Output (Protobuf Lease "leaseKeepAlive")))
IO (NextElem (Proto LeaseKeepAliveResponse))
recv IO (NextElem (Proto LeaseKeepAliveResponse))
-> (NextElem (Proto LeaseKeepAliveResponse) -> IO Int64)
-> IO Int64
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          NextElem Proto LeaseKeepAliveResponse
res -> Int64 -> IO Int64
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int64 -> IO Int64) -> Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ Proto LeaseKeepAliveResponse
res Proto LeaseKeepAliveResponse
-> Getting Int64 (Proto LeaseKeepAliveResponse) Int64 -> Int64
forall s a. s -> Getting a s a -> a
^. Getting Int64 (Proto LeaseKeepAliveResponse) Int64
#ttl
          NextElem (Proto LeaseKeepAliveResponse)
NoNextElem -> do
            Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer EtcdLog
NoKeepAliveResponse
            Int64 -> IO Int64
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int64
0

  writeAlive :: Int64 -> IO ()
writeAlive Int64
leaseId = Text -> IO () -> IO ()
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"writeAlive" (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    IO (Proto PutResponse) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Proto PutResponse) -> IO ())
-> (Proto PutRequest -> IO (Proto PutResponse))
-> Proto PutRequest
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf KV "put")
-> Input (Protobuf KV "put")
-> IO (Output (Protobuf KV "put"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf KV "put")) (Proto PutRequest -> IO ()) -> Proto PutRequest -> IO ()
forall a b. (a -> b) -> a -> b
$
      Proto PutRequest
forall msg. Message msg => msg
defMessage
        Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#key ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
"alive-" ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> Host -> ByteString
forall b a. (Show a, IsString b) => a -> b
show Host
advertise
        Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
#value ASetter (Proto PutRequest) (Proto PutRequest) ByteString ByteString
-> ByteString -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Host -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' Host
advertise
        Proto PutRequest
-> (Proto PutRequest -> Proto PutRequest) -> Proto PutRequest
forall a b. a -> (a -> b) -> b
& ASetter (Proto PutRequest) (Proto PutRequest) Int64 Int64
#lease ASetter (Proto PutRequest) (Proto PutRequest) Int64 Int64
-> Int64 -> Proto PutRequest -> Proto PutRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Int64
leaseId

  getAlive :: IO [Host]
getAlive = Text -> IO [Host] -> IO [Host]
forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
"getAlive" (IO [Host] -> IO [Host]) -> IO [Host] -> IO [Host]
forall a b. (a -> b) -> a -> b
$ do
    Proto RangeResponse
res <-
      Connection
-> ClientHandler'
     'NonStreaming (ReaderT Connection IO) (Protobuf KV "range")
-> Input (Protobuf KV "range")
-> IO (Output (Protobuf KV "range"))
forall {k} (rpc :: k) (m :: * -> *).
Connection
-> ClientHandler' 'NonStreaming (ReaderT Connection m) rpc
-> Input rpc
-> m (Output rpc)
nonStreaming Connection
conn (forall {k} (rpc :: k) (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
forall rpc (styp :: StreamingType) (m :: * -> *).
(CanCallRPC m, SupportsClientRpc rpc,
 SupportsStreamingType rpc styp, Default (RequestMetadata rpc)) =>
ClientHandler' styp m rpc
rpc @(Protobuf KV "range")) (Input (Protobuf KV "range") -> IO (Output (Protobuf KV "range")))
-> Input (Protobuf KV "range") -> IO (Output (Protobuf KV "range"))
forall a b. (a -> b) -> a -> b
$
        Proto RangeRequest
forall msg. Message msg => msg
defMessage
          Proto RangeRequest
-> (Proto RangeRequest -> Proto RangeRequest) -> Proto RangeRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
#key ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
-> ByteString -> Proto RangeRequest -> Proto RangeRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
"alive"
          Proto RangeRequest
-> (Proto RangeRequest -> Proto RangeRequest) -> Proto RangeRequest
forall a b. a -> (a -> b) -> b
& ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
#rangeEnd ASetter
  (Proto RangeRequest) (Proto RangeRequest) ByteString ByteString
-> ByteString -> Proto RangeRequest -> Proto RangeRequest
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ByteString
"alivf" -- NOTE: e+1 to query prefixes
    ((Proto KeyValue -> IO (Maybe Host))
 -> [Proto KeyValue] -> IO [Host])
-> [Proto KeyValue]
-> (Proto KeyValue -> IO (Maybe Host))
-> IO [Host]
forall a b c. (a -> b -> c) -> b -> a -> c
flip (Proto KeyValue -> IO (Maybe Host))
-> [Proto KeyValue] -> IO [Host]
forall (m :: * -> *) a b.
Monad m =>
(a -> m (Maybe b)) -> [a] -> m [b]
mapMaybeM (Proto RangeResponse
res Proto RangeResponse
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> Proto RangeResponse
    -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
-> [Proto KeyValue]
forall s a. s -> Getting (Endo [a]) s a -> [a]
^.. ([Proto KeyValue]
 -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
-> Proto RangeResponse
-> Const (Endo [Proto KeyValue]) (Proto RangeResponse)
#kvs (([Proto KeyValue]
  -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
 -> Proto RangeResponse
 -> Const (Endo [Proto KeyValue]) (Proto RangeResponse))
-> ((Proto KeyValue
     -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
    -> [Proto KeyValue]
    -> Const (Endo [Proto KeyValue]) [Proto KeyValue])
-> (Proto KeyValue
    -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> Proto RangeResponse
-> Const (Endo [Proto KeyValue]) (Proto RangeResponse)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Proto KeyValue -> Const (Endo [Proto KeyValue]) (Proto KeyValue))
-> [Proto KeyValue]
-> Const (Endo [Proto KeyValue]) [Proto KeyValue]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse) ((Proto KeyValue -> IO (Maybe Host)) -> IO [Host])
-> (Proto KeyValue -> IO (Maybe Host)) -> IO [Host]
forall a b. (a -> b) -> a -> b
$ \Proto KeyValue
kv -> do
      let value :: ByteString
value = Proto KeyValue
kv Proto KeyValue
-> Getting ByteString (Proto KeyValue) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto KeyValue) ByteString
#value
      case ByteString -> Either DecoderError Host
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' ByteString
value of
        Left DecoderError
err -> do
          Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith
            Tracer IO EtcdLog
tracer
            FailedToDecodeValue
              { $sel:key:EtcdLog :: Text
key = ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 (ByteString -> Text) -> ByteString -> Text
forall a b. (a -> b) -> a -> b
$ Proto KeyValue
kv Proto KeyValue
-> Getting ByteString (Proto KeyValue) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto KeyValue) ByteString
#key
              , $sel:value:EtcdLog :: Text
value = ByteString -> Text
encodeBase16 ByteString
value
              , $sel:reason:EtcdLog :: Text
reason = DecoderError -> Text
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err
              }
          Maybe Host -> IO (Maybe Host)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe Host
forall a. Maybe a
Nothing
        Right Host
x -> Maybe Host -> IO (Maybe Host)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe Host -> IO (Maybe Host)) -> Maybe Host -> IO (Maybe Host)
forall a b. (a -> b) -> a -> b
$ Host -> Maybe Host
forall a. a -> Maybe a
Just Host
x

-- | Predicate for gRPC errors that we treat as transient — connection blips
-- or etcd-side disruption from which a retry is expected to recover. Anything
-- outside this set is escalated by re-throwing.
--
-- 'GrpcNotFound' is included specifically for lease loss: when etcd's RAFT
-- leader changes under network stress, in-flight leases are revoked, and the
-- next operation referencing one ('pollConnectivity.writeAlive') comes back
-- with NOT_FOUND. The recovery is to mark the network as disconnected and let
-- the outer loop recreate the lease, not crash the node.
isTransientGrpcError :: GrpcError -> Bool
isTransientGrpcError :: GrpcError -> Bool
isTransientGrpcError =
  (GrpcError -> [GrpcError] -> Bool
forall (f :: * -> *) a.
(Foldable f, DisallowElem f, Eq a) =>
a -> f a -> Bool
`elem` [GrpcError
GrpcUnavailable, GrpcError
GrpcDeadlineExceeded, GrpcError
GrpcCancelled, GrpcError
GrpcNotFound])

-- | Classify a failure talking to our local etcd, giving a reason to trace when
-- retrying is the right response.
--
-- Connection-level failures qualify: etcd is a subprocess we started and
-- 'connParams' reconnects, so a lost connection is a blip rather than a reason
-- to take the node down. Those arrive either as 'HTTP2Error' (straight from the
-- @http2@ client, e.g. the SETTINGS rate limit of #2817) or 'ServerDisconnected'
-- (grapesy's wrapper when a call outlives its connection). Everything else
-- escalates, including 'putMessage's cluster-reset 'fail' and
-- 'LastKnownRevisionException'.
retryableEtcdError :: SomeException -> Maybe Text
retryableEtcdError :: SomeException -> Maybe Text
retryableEtcdError SomeException
e
  | Just GrpcException{GrpcError
grpcError :: GrpcError
grpcError :: GrpcException -> GrpcError
grpcError, Maybe Text
grpcErrorMessage :: Maybe Text
grpcErrorMessage :: GrpcException -> Maybe Text
grpcErrorMessage} <- SomeException -> Maybe GrpcException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e =
      if GrpcError -> Bool
isTransientGrpcError GrpcError
grpcError
        then Text -> Maybe Text
forall a. a -> Maybe a
Just (Text -> Maybe Text) -> Text -> Maybe Text
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> Text
forall a. a -> Maybe a -> a
fromMaybe (GrpcError -> Text
forall b a. (Show a, IsString b) => a -> b
show GrpcError
grpcError) Maybe Text
grpcErrorMessage
        else Maybe Text
forall a. Maybe a
Nothing
  | Just (HTTP2Error
http2Error :: HTTP2Error) <- SomeException -> Maybe HTTP2Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e = Text -> Maybe Text
forall a. a -> Maybe a
Just (Text -> Maybe Text) -> Text -> Maybe Text
forall a b. (a -> b) -> a -> b
$ HTTP2Error -> Text
forall b a. (Show a, IsString b) => a -> b
show HTTP2Error
http2Error
  | Just (ServerDisconnected
disconnected :: ServerDisconnected) <- SomeException -> Maybe ServerDisconnected
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e = Text -> Maybe Text
forall a. a -> Maybe a
Just (Text -> Maybe Text) -> Text -> Maybe Text
forall a b. (a -> b) -> a -> b
$ ServerDisconnected -> Text
forall b a. (Show a, IsString b) => a -> b
show ServerDisconnected
disconnected
  | Bool
otherwise = Maybe Text
forall a. Maybe a
Nothing

-- | Add context to the 'grpcErrorMessage' of any 'GrpcException' raised.
withGrpcContext :: MonadCatch m => Text -> m a -> m a
withGrpcContext :: forall (m :: * -> *) a. MonadCatch m => Text -> m a -> m a
withGrpcContext Text
context m a
action =
  m a
action m a -> (GrpcException -> m a) -> m a
forall e a. Exception e => m a -> (e -> m a) -> m a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \e :: GrpcException
e@GrpcException{Maybe Text
grpcErrorMessage :: GrpcException -> Maybe Text
grpcErrorMessage :: Maybe Text
grpcErrorMessage} ->
    GrpcException -> m a
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO
      GrpcException
e
        { grpcErrorMessage =
            case grpcErrorMessage of
              Maybe Text
Nothing -> Text -> Maybe Text
forall a. a -> Maybe a
Just Text
context
              Just Text
msg -> Text -> Maybe Text
forall a. a -> Maybe a
Just (Text -> Maybe Text) -> Text -> Maybe Text
forall a b. (a -> b) -> a -> b
$ Text
context Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
": " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
msg
        }

-- | Like 'withProcessTerm', but sends first SIGINT and only SIGTERM if not
-- stopped within 5 seconds.
withProcessInterrupt ::
  (MonadIO m, MonadThrow m) =>
  ProcessConfig stdin stdout stderr ->
  (Process stdin stdout stderr -> m a) ->
  m a
withProcessInterrupt :: forall (m :: * -> *) stdin stdout stderr a.
(MonadIO m, MonadThrow m) =>
ProcessConfig stdin stdout stderr
-> (Process stdin stdout stderr -> m a) -> m a
withProcessInterrupt ProcessConfig stdin stdout stderr
config =
  m (Process stdin stdout stderr)
-> (Process stdin stdout stderr -> m ())
-> (Process stdin stdout stderr -> m a)
-> m a
forall a b c. m a -> (a -> m b) -> (a -> m c) -> m c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (ProcessConfig stdin stdout stderr
-> m (Process stdin stdout stderr)
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
ProcessConfig stdin stdout stderr
-> m (Process stdin stdout stderr)
startProcess (ProcessConfig stdin stdout stderr
 -> m (Process stdin stdout stderr))
-> ProcessConfig stdin stdout stderr
-> m (Process stdin stdout stderr)
forall a b. (a -> b) -> a -> b
$ ProcessConfig stdin stdout stderr
config ProcessConfig stdin stdout stderr
-> (ProcessConfig stdin stdout stderr
    -> ProcessConfig stdin stdout stderr)
-> ProcessConfig stdin stdout stderr
forall a b. a -> (a -> b) -> b
& Bool
-> ProcessConfig stdin stdout stderr
-> ProcessConfig stdin stdout stderr
forall stdin stdout stderr.
Bool
-> ProcessConfig stdin stdout stderr
-> ProcessConfig stdin stdout stderr
setCreateGroup Bool
True) Process stdin stdout stderr -> m ()
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
Process stdin stdout stderr -> m ()
signalAndStopProcess
 where
  signalAndStopProcess :: MonadIO m => Process stdin stdout stderr -> m ()
  signalAndStopProcess :: forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
Process stdin stdout stderr -> m ()
signalAndStopProcess Process stdin stdout stderr
p = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    ProcessHandle -> IO ()
interruptProcessGroupOf (Process stdin stdout stderr -> ProcessHandle
forall stdin stdout stderr.
Process stdin stdout stderr -> ProcessHandle
unsafeProcessHandle Process stdin stdout stderr
p)
    (String, IO ()) -> (String, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (String, m b) -> m ()
raceLabelled_
      (String
"etcd-signalAndStopProcess-waitExitCode", IO ExitCode -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO ExitCode -> IO ()) -> IO ExitCode -> IO ()
forall a b. (a -> b) -> a -> b
$ Process stdin stdout stderr -> IO ExitCode
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
Process stdin stdout stderr -> m ExitCode
waitExitCode Process stdin stdout stderr
p)
      -- 'stopProcess' can lose a reap race against its own waiter thread when
      -- the process dies right around the SIGTERM escalation: its internal
      -- 'waitForProcess' then throws ECHILD ("does not exist"). The process is
      -- dead either way, which is all we wanted here.
      ( String
"etcd-signalAndStopProcess-stopProcess"
      , DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
5
          IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Process stdin stdout stderr -> IO ()
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
Process stdin stdout stderr -> m ()
stopProcess Process stdin stdout stderr
p
          IO () -> (IOError -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` (\IOError
e -> Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (IOError -> Bool
isDoesNotExistError IOError
e) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ IOError -> IO ()
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO IOError
e)
      )

-- * Persistent queue

-- | Queue elements carry the item's CBOR serialization, produced once on
-- write and reused for both the on-disk file and the etcd value, so
-- broadcasting does not serialize twice.
data PersistentQueue m a = PersistentQueue
  { forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
  , forall (m :: * -> *) a. PersistentQueue m a -> TVar m Natural
nextIx :: TVar m Natural
  , forall (m :: * -> *) a. PersistentQueue m a -> String
directory :: FilePath
  }

-- | Create a new persistent queue at file path and given capacity.
newPersistentQueue ::
  (MonadLabelledSTM m, MonadIO m, FromCBOR a, MonadCatch m, MonadFail m) =>
  Tracer IO EtcdLog ->
  FilePath ->
  Natural ->
  m (PersistentQueue m a)
newPersistentQueue :: forall (m :: * -> *) a.
(MonadLabelledSTM m, MonadIO m, FromCBOR a, MonadCatch m,
 MonadFail m) =>
Tracer IO EtcdLog -> String -> Natural -> m (PersistentQueue m a)
newPersistentQueue Tracer IO EtcdLog
tracer String
path Natural
capacity = do
  [Natural]
paths <- IO [Natural] -> m [Natural]
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO [Natural] -> m [Natural]) -> IO [Natural] -> m [Natural]
forall a b. (a -> b) -> a -> b
$ do
    Bool -> String -> IO ()
createDirectoryIfMissing Bool
True String
path
    [Natural] -> [Natural]
forall a. Ord a => [a] -> [a]
sort ([Natural] -> [Natural])
-> ([String] -> [Natural]) -> [String] -> [Natural]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (String -> Maybe Natural) -> [String] -> [Natural]
forall a b. (a -> Maybe b) -> [a] -> [b]
mapMaybe String -> Maybe Natural
forall a. Read a => String -> Maybe a
readMaybe ([String] -> [Natural]) -> IO [String] -> IO [Natural]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> String -> IO [String]
listDirectory String
path
  TBQueue m (Natural, a, ByteString)
queue <- String -> Natural -> m (TBQueue m (Natural, a, ByteString))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> Natural -> m (TBQueue m a)
newLabelledTBQueueIO String
"persistent-queue" (Natural -> m (TBQueue m (Natural, a, ByteString)))
-> Natural -> m (TBQueue m (Natural, a, ByteString))
forall a b. (a -> b) -> a -> b
$ Natural -> Natural -> Natural
forall a. Ord a => a -> a -> a
max (Int -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Natural) -> Int -> Natural
forall a b. (a -> b) -> a -> b
$ [Natural] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Natural]
paths) Natural
capacity
  Natural
highestId <-
    m Natural -> m (Either IOError Natural)
forall e a. Exception e => m a -> m (Either e a)
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try (TBQueue m (Natural, a, ByteString) -> [Natural] -> m Natural
loadExisting TBQueue m (Natural, a, ByteString)
queue [Natural]
paths) m (Either IOError Natural)
-> (Either IOError Natural -> m Natural) -> m Natural
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Left (IOError
e :: IOException) -> do
        IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ do
          Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer PersistentQueueLoadFailed{$sel:reason:EtcdLog :: Text
reason = IOError -> Text
forall b a. (Show a, IsString b) => a -> b
show IOError
e}
          Bool -> String -> IO ()
createDirectoryIfMissing Bool
True String
path
        Natural -> m Natural
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
0
      Right Natural
highest -> Natural -> m Natural
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
highest
  TVar m Natural
nextIx <- String -> Natural -> m (TVar m Natural)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"persistent-next-ix" (Natural -> m (TVar m Natural)) -> Natural -> m (TVar m Natural)
forall a b. (a -> b) -> a -> b
$ Natural
highestId Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
1
  PersistentQueue m a -> m (PersistentQueue m a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue, TVar m Natural
$sel:nextIx:PersistentQueue :: TVar m Natural
nextIx :: TVar m Natural
nextIx, $sel:directory:PersistentQueue :: String
directory = String
path}
 where
  loadExisting :: TBQueue m (Natural, a, ByteString) -> [Natural] -> m Natural
loadExisting TBQueue m (Natural, a, ByteString)
queue = \case
    [] -> Natural -> m Natural
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
0
    [Natural]
idxs -> do
      [Natural] -> (Natural -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Natural]
idxs ((Natural -> m ()) -> m ()) -> (Natural -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \(Natural
idx :: Natural) -> do
        ByteString
bs <- String -> m ByteString
forall (m :: * -> *). MonadIO m => String -> m ByteString
readFileBS (String
path String -> String -> String
</> Natural -> String
forall b a. (Show a, IsString b) => a -> b
show Natural
idx)
        case ByteString -> Either DecoderError a
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' ByteString
bs of
          Left DecoderError
err ->
            String -> m ()
forall a. String -> m a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail (String -> m ()) -> String -> m ()
forall a b. (a -> b) -> a -> b
$ String
"Failed to decode item: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> DecoderError -> String
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err
          Right a
item ->
            STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TBQueue m (Natural, a, ByteString)
-> (Natural, a, ByteString) -> STM m ()
forall a. TBQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue m (Natural, a, ByteString)
queue (Natural
idx, a
item, ByteString
bs)
      Natural -> m Natural
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Natural -> m Natural) -> Natural -> m Natural
forall a b. (a -> b) -> a -> b
$ [Natural] -> Natural
forall a. HasCallStack => [a] -> a
List.last [Natural]
idxs

-- | Write a value to the queue, blocking if the queue is full.
writePersistentQueue :: (ToCBOR a, MonadSTM m, MonadIO m) => Tracer IO EtcdLog -> PersistentQueue m a -> a -> m ()
writePersistentQueue :: forall a (m :: * -> *).
(ToCBOR a, MonadSTM m, MonadIO m) =>
Tracer IO EtcdLog -> PersistentQueue m a -> a -> m ()
writePersistentQueue Tracer IO EtcdLog
tracer PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue, TVar m Natural
$sel:nextIx:PersistentQueue :: forall (m :: * -> *) a. PersistentQueue m a -> TVar m Natural
nextIx :: TVar m Natural
nextIx, String
$sel:directory:PersistentQueue :: forall (m :: * -> *) a. PersistentQueue m a -> String
directory :: String
directory} a
item = do
  Natural
next <- STM m Natural -> m Natural
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m Natural -> m Natural) -> STM m Natural -> m Natural
forall a b. (a -> b) -> a -> b
$ do
    Natural
next <- TVar m Natural -> STM m Natural
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m Natural
nextIx
    TVar m Natural -> (Natural -> Natural) -> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m Natural
nextIx (Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
1)
    Natural -> STM m Natural
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Natural
next
  let !bytes :: ByteString
bytes = a -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' a
item
  String -> ByteString -> m ()
forall (m :: * -> *). MonadIO m => String -> ByteString -> m ()
writeFileBS (String
directory String -> String -> String
</> Natural -> String
forall b a. (Show a, IsString b) => a -> b
show Natural
next) ByteString
bytes
  Bool
full <- STM m Bool -> m Bool
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m Bool -> m Bool) -> STM m Bool -> m Bool
forall a b. (a -> b) -> a -> b
$ TBQueue m (Natural, a, ByteString) -> STM m Bool
forall a. TBQueue m a -> STM m Bool
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m Bool
isFullTBQueue TBQueue m (Natural, a, ByteString)
queue
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
full (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer EtcdLog
PersistentQueueFull
  STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TBQueue m (Natural, a, ByteString)
-> (Natural, a, ByteString) -> STM m ()
forall a. TBQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue m (Natural, a, ByteString)
queue (Natural
next, a
item, ByteString
bytes)

-- | Get the next value from the queue without removing it, blocking if the
-- queue is empty.
peekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m a
peekPersistentQueue :: forall (m :: * -> *) a. MonadSTM m => PersistentQueue m a -> m a
peekPersistentQueue PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue} = do
  (\(Natural
_, a
item, ByteString
_) -> a
item) ((Natural, a, ByteString) -> a)
-> m (Natural, a, ByteString) -> m a
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM m (Natural, a, ByteString) -> m (Natural, a, ByteString)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (TBQueue m (Natural, a, ByteString)
-> STM m (Natural, a, ByteString)
forall a. TBQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m a
peekTBQueue TBQueue m (Natural, a, ByteString)
queue)

-- | Like 'peekPersistentQueue', but returns 'Nothing' instead of blocking
-- when the queue is empty.
tryPeekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m (Maybe a)
tryPeekPersistentQueue :: forall (m :: * -> *) a.
MonadSTM m =>
PersistentQueue m a -> m (Maybe a)
tryPeekPersistentQueue PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue} = do
  ((Natural, a, ByteString) -> a)
-> Maybe (Natural, a, ByteString) -> Maybe a
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (\(Natural
_, a
item, ByteString
_) -> a
item) (Maybe (Natural, a, ByteString) -> Maybe a)
-> m (Maybe (Natural, a, ByteString)) -> m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM m (Maybe (Natural, a, ByteString))
-> m (Maybe (Natural, a, ByteString))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (TBQueue m (Natural, a, ByteString)
-> STM m (Maybe (Natural, a, ByteString))
forall a. TBQueue m a -> STM m (Maybe a)
forall (m :: * -> *) a.
MonadSTM m =>
TBQueue m a -> STM m (Maybe a)
tryPeekTBQueue TBQueue m (Natural, a, ByteString)
queue)

-- | Get all pending values and their serializations, up to the given count
-- and total byte limits, blocking until at least one is available. Values
-- are not removed; use 'popBatchPersistentQueue' after they were sent.
peekBatchPersistentQueue :: MonadSTM m => PersistentQueue m a -> Int -> Int -> m [(a, ByteString)]
peekBatchPersistentQueue :: forall (m :: * -> *) a.
MonadSTM m =>
PersistentQueue m a -> Int -> Int -> m [(a, ByteString)]
peekBatchPersistentQueue PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue} Int
maxCount Int
maxBytes = STM m [(a, ByteString)] -> m [(a, ByteString)]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [(a, ByteString)] -> m [(a, ByteString)])
-> STM m [(a, ByteString)] -> m [(a, ByteString)]
forall a b. (a -> b) -> a -> b
$ do
  (Natural, a, ByteString)
first' <- TBQueue m (Natural, a, ByteString)
-> STM m (Natural, a, ByteString)
forall a. TBQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m a
readTBQueue TBQueue m (Natural, a, ByteString)
queue
  -- Collected in reverse consumption order
  [(Natural, a, ByteString)]
rest <- Int
-> [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
go ((Natural, a, ByteString) -> Int
remainingAfter (Natural, a, ByteString)
first') []
  -- Restore everything we consumed: 'unGetTBQueue' pushes to the front, so
  -- restoring newest-first re-establishes the original queue order.
  [(Natural, a, ByteString)]
-> ((Natural, a, ByteString) -> STM m ()) -> STM m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ ([(Natural, a, ByteString)]
rest [(Natural, a, ByteString)]
-> [(Natural, a, ByteString)] -> [(Natural, a, ByteString)]
forall a. Semigroup a => a -> a -> a
<> [(Natural, a, ByteString)
first']) (((Natural, a, ByteString) -> STM m ()) -> STM m ())
-> ((Natural, a, ByteString) -> STM m ()) -> STM m ()
forall a b. (a -> b) -> a -> b
$ TBQueue m (Natural, a, ByteString)
-> (Natural, a, ByteString) -> STM m ()
forall a. TBQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
unGetTBQueue TBQueue m (Natural, a, ByteString)
queue
  [(a, ByteString)] -> STM m [(a, ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([(a, ByteString)] -> STM m [(a, ByteString)])
-> [(a, ByteString)] -> STM m [(a, ByteString)]
forall a b. (a -> b) -> a -> b
$ (\(Natural
_, a
item, ByteString
bytes) -> (a
item, ByteString
bytes)) ((Natural, a, ByteString) -> (a, ByteString))
-> [(Natural, a, ByteString)] -> [(a, ByteString)]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ((Natural, a, ByteString)
first' (Natural, a, ByteString)
-> [(Natural, a, ByteString)] -> [(Natural, a, ByteString)]
forall a. a -> [a] -> [a]
: [(Natural, a, ByteString)] -> [(Natural, a, ByteString)]
forall a. [a] -> [a]
reverse [(Natural, a, ByteString)]
rest)
 where
  go :: Int
-> [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
go Int
budget [(Natural, a, ByteString)]
acc
    | [(Natural, a, ByteString)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(Natural, a, ByteString)]
acc Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
maxCount Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1 = [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [(Natural, a, ByteString)]
acc
    | Bool
otherwise =
        TBQueue m (Natural, a, ByteString)
-> STM m (Maybe (Natural, a, ByteString))
forall a. TBQueue m a -> STM m (Maybe a)
forall (m :: * -> *) a.
MonadSTM m =>
TBQueue m a -> STM m (Maybe a)
tryReadTBQueue TBQueue m (Natural, a, ByteString)
queue STM m (Maybe (Natural, a, ByteString))
-> (Maybe (Natural, a, ByteString)
    -> STM m [(Natural, a, ByteString)])
-> STM m [(Natural, a, ByteString)]
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          Maybe (Natural, a, ByteString)
Nothing -> [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [(Natural, a, ByteString)]
acc
          Just next :: (Natural, a, ByteString)
next@(Natural
_, a
_, ByteString
bytes)
            | ByteString -> Int
BS.length ByteString
bytes Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
budget -> do
                TBQueue m (Natural, a, ByteString)
-> (Natural, a, ByteString) -> STM m ()
forall a. TBQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
unGetTBQueue TBQueue m (Natural, a, ByteString)
queue (Natural, a, ByteString)
next
                [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [(Natural, a, ByteString)]
acc
            | Bool
otherwise -> Int
-> [(Natural, a, ByteString)] -> STM m [(Natural, a, ByteString)]
go (Int
budget Int -> Int -> Int
forall a. Num a => a -> a -> a
- ByteString -> Int
BS.length ByteString
bytes) ((Natural, a, ByteString)
next (Natural, a, ByteString)
-> [(Natural, a, ByteString)] -> [(Natural, a, ByteString)]
forall a. a -> [a] -> [a]
: [(Natural, a, ByteString)]
acc)

  remainingAfter :: (Natural, a, ByteString) -> Int
remainingAfter (Natural
_, a
_, ByteString
bytes) = Int
maxBytes Int -> Int -> Int
forall a. Num a => a -> a -> a
- ByteString -> Int
BS.length ByteString
bytes

-- | Get the batch to broadcast next: the batch already in flight if there is
-- one, otherwise a fresh one peeked from the queue (and recorded as in
-- flight). Returns 'Nothing' when nothing is pending. The caller must clear
-- the in-flight var after popping a successfully sent batch.
--
-- Pinning the in-flight batch across transient retries matters for the
-- compare-fail dedup in 'putMessage': a put can commit server-side while the
-- client sees e.g. 'GrpcDeadlineExceeded'. The retry must send (and
-- afterwards pop) exactly the content of the committed attempt. Re-peeking
-- on retry could pick up messages enqueued in the meantime; the dedup branch
-- would then declare the grown batch delivered and the never-sent tail would
-- be popped and lost.
nextPendingBatch ::
  MonadSTM m =>
  TVar m (Maybe [(a, ByteString)]) ->
  PersistentQueue m a ->
  Int ->
  Int ->
  m (Maybe [(a, ByteString)])
nextPendingBatch :: forall (m :: * -> *) a.
MonadSTM m =>
TVar m (Maybe [(a, ByteString)])
-> PersistentQueue m a -> Int -> Int -> m (Maybe [(a, ByteString)])
nextPendingBatch TVar m (Maybe [(a, ByteString)])
inFlightVar PersistentQueue m a
queue Int
maxCount Int
maxBytes =
  TVar m (Maybe [(a, ByteString)]) -> m (Maybe [(a, ByteString)])
forall a. TVar m a -> m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar m (Maybe [(a, ByteString)])
inFlightVar m (Maybe [(a, ByteString)])
-> (Maybe [(a, ByteString)] -> m (Maybe [(a, ByteString)]))
-> m (Maybe [(a, ByteString)])
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
    Just [(a, ByteString)]
batch -> Maybe [(a, ByteString)] -> m (Maybe [(a, ByteString)])
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([(a, ByteString)] -> Maybe [(a, ByteString)]
forall a. a -> Maybe a
Just [(a, ByteString)]
batch)
    Maybe [(a, ByteString)]
Nothing ->
      PersistentQueue m a -> m (Maybe a)
forall (m :: * -> *) a.
MonadSTM m =>
PersistentQueue m a -> m (Maybe a)
tryPeekPersistentQueue PersistentQueue m a
queue m (Maybe a)
-> (Maybe a -> m (Maybe [(a, ByteString)]))
-> m (Maybe [(a, ByteString)])
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Maybe a
Nothing -> Maybe [(a, ByteString)] -> m (Maybe [(a, ByteString)])
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe [(a, ByteString)]
forall a. Maybe a
Nothing
        Just a
_ -> do
          -- All pending messages (bounded by the given limits) are sent as a
          -- single etcd value, so a whole batch costs one Raft commit
          -- instead of one per message.
          [(a, ByteString)]
batch <- PersistentQueue m a -> Int -> Int -> m [(a, ByteString)]
forall (m :: * -> *) a.
MonadSTM m =>
PersistentQueue m a -> Int -> Int -> m [(a, ByteString)]
peekBatchPersistentQueue PersistentQueue m a
queue Int
maxCount Int
maxBytes
          STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Maybe [(a, ByteString)])
-> Maybe [(a, ByteString)] -> STM m ()
forall a. TVar m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar m (Maybe [(a, ByteString)])
inFlightVar ([(a, ByteString)] -> Maybe [(a, ByteString)]
forall a. a -> Maybe a
Just [(a, ByteString)]
batch)
          Maybe [(a, ByteString)] -> m (Maybe [(a, ByteString)])
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([(a, ByteString)] -> Maybe [(a, ByteString)]
forall a. a -> Maybe a
Just [(a, ByteString)]
batch)

-- | Remove a batch previously returned by 'peekBatchPersistentQueue'. Pops
-- unconditionally, one item per batch entry: this thread is the sole
-- consumer, so the queue head still holds exactly the peeked items (an
-- item-matching guard could only silently no-op and wedge the queue, see
-- #2742).
popBatchPersistentQueue :: (MonadSTM m, MonadIO m) => Tracer IO EtcdLog -> PersistentQueue m a -> [(a, ByteString)] -> m ()
popBatchPersistentQueue :: forall (m :: * -> *) a.
(MonadSTM m, MonadIO m) =>
Tracer IO EtcdLog
-> PersistentQueue m a -> [(a, ByteString)] -> m ()
popBatchPersistentQueue Tracer IO EtcdLog
tracer PersistentQueue{TBQueue m (Natural, a, ByteString)
$sel:queue:PersistentQueue :: forall (m :: * -> *) a.
PersistentQueue m a -> TBQueue m (Natural, a, ByteString)
queue :: TBQueue m (Natural, a, ByteString)
queue, String
$sel:directory:PersistentQueue :: forall (m :: * -> *) a. PersistentQueue m a -> String
directory :: String
directory} [(a, ByteString)]
batch = do
  [Natural]
indices <- STM m [Natural] -> m [Natural]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [Natural] -> m [Natural]) -> STM m [Natural] -> m [Natural]
forall a b. (a -> b) -> a -> b
$ [(a, ByteString)]
-> ((a, ByteString) -> STM m Natural) -> STM m [Natural]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [(a, ByteString)]
batch (((a, ByteString) -> STM m Natural) -> STM m [Natural])
-> ((a, ByteString) -> STM m Natural) -> STM m [Natural]
forall a b. (a -> b) -> a -> b
$ \(a, ByteString)
_ -> (\(Natural
ix, a
_, ByteString
_) -> Natural
ix) ((Natural, a, ByteString) -> Natural)
-> STM m (Natural, a, ByteString) -> STM m Natural
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TBQueue m (Natural, a, ByteString)
-> STM m (Natural, a, ByteString)
forall a. TBQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m a
readTBQueue TBQueue m (Natural, a, ByteString)
queue
  [Natural] -> (Natural -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Natural]
indices ((Natural -> m ()) -> m ()) -> (Natural -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ Tracer IO EtcdLog -> String -> Natural -> m ()
forall (m :: * -> *).
MonadIO m =>
Tracer IO EtcdLog -> String -> Natural -> m ()
removeQueueFile Tracer IO EtcdLog
tracer String
directory

-- | Delete the backing file of a popped queue item. Failing to delete is
-- traced but not fatal: the message was already broadcast, so a leftover
-- file only means it may be re-broadcast after a restart (at-least-once
-- delivery, same as the crash-recovery path).
removeQueueFile :: MonadIO m => Tracer IO EtcdLog -> FilePath -> Natural -> m ()
removeQueueFile :: forall (m :: * -> *).
MonadIO m =>
Tracer IO EtcdLog -> String -> Natural -> m ()
removeQueueFile Tracer IO EtcdLog
tracer String
directory Natural
ix =
  IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$
    String -> IO ()
removeFile (String
directory String -> String -> String
</> Natural -> String
forall b a. (Show a, IsString b) => a -> b
show Natural
ix) IO () -> (IOError -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \IOError
e ->
      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (IOError -> Bool
isDoesNotExistError IOError
e) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        Tracer IO EtcdLog -> EtcdLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO EtcdLog
tracer PersistentQueueDeleteFailed{$sel:index:EtcdLog :: Natural
index = Natural
ix, $sel:reason:EtcdLog :: Text
reason = IOError -> Text
forall b a. (Show a, IsString b) => a -> b
show IOError
e}

-- * Tracing

data EtcdLog
  = EtcdLog {EtcdLog -> Value
etcd :: Value}
  | Reconnecting
  | BroadcastFailed {EtcdLog -> Text
reason :: Text}
  | FailedToDecodeLog {EtcdLog -> Text
log :: Text, reason :: Text}
  | FailedToDecodeValue {EtcdLog -> Text
key :: Text, EtcdLog -> Text
value :: Text, reason :: Text}
  | CreatedLease {EtcdLog -> Int64
leaseId :: Int64}
  | LowLeaseTTL {EtcdLog -> Int64
ttlRemaining :: Int64}
  | NoKeepAliveResponse
  | MatchingProtocolVersion {EtcdLog -> ProtocolVersion
version :: ProtocolVersion}
  | WatchMessagesStartRevision {EtcdLog -> Int64
startRevision :: Int64}
  | WatchMessagesFallbackTo {EtcdLog -> Int64
compactRevision :: Int64}
  | -- | The watch stream failed; it is restarted from the last known revision.
    WatchFailed {reason :: Text}
  | -- | The etcd transaction wrapping a broadcast 'put' found that our
    -- key's @mod_revision@ had moved past what we last recorded — the
    -- expected outcome when a 'GrpcDeadlineExceeded'-retried put already
    -- committed server-side. No second put was issued.
    BroadcastDeduped {EtcdLog -> Int64
previousModRev :: Int64, EtcdLog -> Int64
observedModRev :: Int64}
  | -- | Failed to load persisted queue items from disk on startup. The queue
    -- starts empty; any in-flight messages from before the crash are lost.
    PersistentQueueLoadFailed {reason :: Text}
  | -- | The persistent queue has reached capacity. The calling thread will
    -- block until the broadcast loop drains at least one item.
    PersistentQueueFull
  | -- | Failed to delete the backing file of an already-broadcast item. The
    -- queue keeps operating; the leftover file only means the message may be
    -- re-broadcast after a restart.
    PersistentQueueDeleteFailed {EtcdLog -> Natural
index :: Natural, reason :: Text}
  deriving stock (EtcdLog -> EtcdLog -> Bool
(EtcdLog -> EtcdLog -> Bool)
-> (EtcdLog -> EtcdLog -> Bool) -> Eq EtcdLog
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: EtcdLog -> EtcdLog -> Bool
== :: EtcdLog -> EtcdLog -> Bool
$c/= :: EtcdLog -> EtcdLog -> Bool
/= :: EtcdLog -> EtcdLog -> Bool
Eq, Int -> EtcdLog -> String -> String
[EtcdLog] -> String -> String
EtcdLog -> String
(Int -> EtcdLog -> String -> String)
-> (EtcdLog -> String)
-> ([EtcdLog] -> String -> String)
-> Show EtcdLog
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> EtcdLog -> String -> String
showsPrec :: Int -> EtcdLog -> String -> String
$cshow :: EtcdLog -> String
show :: EtcdLog -> String
$cshowList :: [EtcdLog] -> String -> String
showList :: [EtcdLog] -> String -> String
Show, (forall x. EtcdLog -> Rep EtcdLog x)
-> (forall x. Rep EtcdLog x -> EtcdLog) -> Generic EtcdLog
forall x. Rep EtcdLog x -> EtcdLog
forall x. EtcdLog -> Rep EtcdLog x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. EtcdLog -> Rep EtcdLog x
from :: forall x. EtcdLog -> Rep EtcdLog x
$cto :: forall x. Rep EtcdLog x -> EtcdLog
to :: forall x. Rep EtcdLog x -> EtcdLog
Generic)
  deriving anyclass ([EtcdLog] -> Value
[EtcdLog] -> Encoding
EtcdLog -> Bool
EtcdLog -> Value
EtcdLog -> Encoding
(EtcdLog -> Value)
-> (EtcdLog -> Encoding)
-> ([EtcdLog] -> Value)
-> ([EtcdLog] -> Encoding)
-> (EtcdLog -> Bool)
-> ToJSON EtcdLog
forall a.
(a -> Value)
-> (a -> Encoding)
-> ([a] -> Value)
-> ([a] -> Encoding)
-> (a -> Bool)
-> ToJSON a
$ctoJSON :: EtcdLog -> Value
toJSON :: EtcdLog -> Value
$ctoEncoding :: EtcdLog -> Encoding
toEncoding :: EtcdLog -> Encoding
$ctoJSONList :: [EtcdLog] -> Value
toJSONList :: [EtcdLog] -> Value
$ctoEncodingList :: [EtcdLog] -> Encoding
toEncodingList :: [EtcdLog] -> Encoding
$comitField :: EtcdLog -> Bool
omitField :: EtcdLog -> Bool
ToJSON)