{-# LANGUAGE OverloadedRecordDot #-}

-- | Test the real networking layer
module Hydra.NetworkSpec where

import Hydra.Prelude
import Test.Hydra.Prelude

import Cardano.Binary (serialize')
import Codec.CBOR.Read (deserialiseFromBytes)
import Codec.CBOR.Write (toLazyByteString)
import Control.Concurrent.Class.MonadSTM (
  modifyTVar',
  readTBQueue,
  readTQueue,
  readTVarIO,
  writeTBQueue,
  writeTQueue,
 )
import Data.Bits (testBit)
import Data.ByteString qualified as BS
import Data.Text qualified as T
import Hydra.Ledger.Simple (SimpleTx (..))
import Hydra.Logging (Envelope (message), showLogsOnFailure, traceInTVar)
import Hydra.Network (
  Connectivity (..),
  Host (..),
  Network (..),
  NetworkCallback (..),
  ProtocolVersion (..),
  WhichEtcd (..),
 )
import Hydra.Network.Etcd (EtcdLog (..), LastKnownRevisionException (..), batchValue, connParams, getClientPort, getLastKnownRevision, grpcServer, isTransientGrpcError, peerPortToClientPort, putLastKnownRevision, putMessage, queryInitialModRev, retryableEtcdError, withEtcdNetwork)
import Hydra.Network.Message (Message (..))
import Hydra.Node.Network (NetworkConfiguration (..))
import Network.GRPC.Client (Address (..), Server (..), ServerDisconnected (..), withConnection)
import Network.GRPC.Common (GrpcError (..), GrpcException (..))
import Network.HTTP2.Client (ErrorCode (..), HTTP2Error (..))
import Network.Socket (
  Family (AF_INET),
  PortNumber,
  SockAddr (SockAddrInet),
  SocketOption (ReuseAddr),
  SocketType (Stream),
  accept,
  bind,
  close,
  connect,
  defaultProtocol,
  setSocketOption,
  socket,
  socketPort,
  tupleToHostAddress,
 )
import Network.Socket qualified as Socket
import Network.Socket.ByteString (recv, sendAll)
import System.Directory (removeFile)
import System.FilePath ((</>))
import System.IO.Error (userError)
import System.Process.Typed (readProcessStdout_, runProcess_, shell)
import Test.Aeson.GenericSpecs (Settings (..), defaultSettings, roundtripAndGoldenADTSpecsWithSettings)
import Test.Hydra.Ledger.Simple ()
import Test.Hydra.Network.Message ()
import Test.Hydra.Node.Fixture (alice, aliceSk, bob, bobSk, carol, carolSk)
import Test.Network.Ports (randomUnusedTCPPortsWithDerived, withFreePortAndDerived)
import Test.QuickCheck (Property, (===))
import Test.QuickCheck.Instances.ByteString ()
import Test.Util (noopCallback, waitEq, waitMatch)

spec :: Spec
spec :: Spec
spec = do
  -- TODO: add tests about advertise being honored
  --
  -- Per-test 'failAfter' budgets in this block are deliberately generous
  -- (60s). The previous 15–30s budgets fired too eagerly when an etcd
  -- election or a gRPC reconnect happened to land on the same scheduler
  -- tick as a CI slowdown. The 'network-test.yaml' workflow (pumba
  -- packet loss) is the operating point this guards against; tighten
  -- only if you've confirmed it still passes there.
  HostName -> Spec -> Spec
forall a. HasCallStack => HostName -> SpecWith a -> SpecWith a
describe HostName
"isTransientGrpcError" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    -- The loss-90 CI failure traced back to 'pollConnectivity.writeAlive'
    -- raising 'GrpcNotFound' when etcd's RAFT leader changed and the lease
    -- was revoked. The error must be classified as transient so the outer
    -- loop can recreate the lease instead of letting the exception escape
    -- and crash the hydra-node process.
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"treats GrpcNotFound as transient (lease loss during leader change)" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcNotFound Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
True
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"still treats the original transient errors as transient" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$ do
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcUnavailable Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
True
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcDeadlineExceeded Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
True
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcCancelled Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
True
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"does not treat unrelated errors as transient" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$ do
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcInvalidArgument Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
False
      GrpcError -> Bool
isTransientGrpcError GrpcError
GrpcPermissionDenied Bool -> Bool -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Bool
False

  HostName -> Spec -> Spec
forall a. HasCallStack => HostName -> SpecWith a -> SpecWith a
describe HostName
"retryableEtcdError" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    -- #2817: the SETTINGS rate limit kills the connection with a bare
    -- 'HTTP2Error', which is not a 'GrpcException' and so escaped every handler
    -- in the module and took the node down.
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"retries http2 connection errors" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
toException (HTTP2Error -> SomeException) -> HTTP2Error -> SomeException
forall a b. (a -> b) -> a -> b
$ ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
ConnectionErrorIsSent ErrorCode
EnhanceYourCalm Int
0 ReasonPhrase
"too many settings")
        Maybe Text -> (Maybe Text -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` Maybe Text -> Bool
forall a. Maybe a -> Bool
isJust
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"retries a connection lost under an in-flight call" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (ServerDisconnected -> SomeException
forall e. Exception e => e -> SomeException
toException (ServerDisconnected -> SomeException)
-> ServerDisconnected -> SomeException
forall a b. (a -> b) -> a -> b
$ SomeException -> CallStack -> ServerDisconnected
ServerDisconnected (HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
toException HTTP2Error
ConnectionIsClosed) CallStack
HasCallStack => CallStack
callStack)
        Maybe Text -> (Maybe Text -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` Maybe Text -> Bool
forall a. Maybe a -> Bool
isJust
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"retries transient grpc errors" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (GrpcException -> SomeException
forall e. Exception e => e -> SomeException
toException GrpcException{grpcError :: GrpcError
grpcError = GrpcError
GrpcUnavailable, grpcErrorMessage :: Maybe Text
grpcErrorMessage = Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"etcd is electing", grpcErrorDetails :: Maybe ByteString
grpcErrorDetails = Maybe ByteString
forall a. Maybe a
Nothing, grpcErrorMetadata :: [CustomMetadata]
grpcErrorMetadata = []})
        Maybe Text -> Maybe Text -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"etcd is electing"
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"escalates other grpc errors" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (GrpcException -> SomeException
forall e. Exception e => e -> SomeException
toException GrpcException{grpcError :: GrpcError
grpcError = GrpcError
GrpcInvalidArgument, grpcErrorMessage :: Maybe Text
grpcErrorMessage = Maybe Text
forall a. Maybe a
Nothing, grpcErrorDetails :: Maybe ByteString
grpcErrorDetails = Maybe ByteString
forall a. Maybe a
Nothing, grpcErrorMetadata :: [CustomMetadata]
grpcErrorMetadata = []})
        Maybe Text -> Maybe Text -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Maybe Text
forall a. Maybe a
Nothing
    -- 'putMessage' fails this way when etcd has lost the key we wrote against;
    -- that has to keep taking the node down.
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"escalates anything else" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (IOError -> SomeException
forall e. Exception e => e -> SomeException
toException (IOError -> SomeException) -> IOError -> SomeException
forall a b. (a -> b) -> a -> b
$ HostName -> IOError
userError HostName
"our broadcast key has no current value in etcd")
        Maybe Text -> Maybe Text -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Maybe Text
forall a. Maybe a
Nothing
    -- Retrying cannot make a bad file readable, so this must escalate rather
    -- than spin in 'waitMessages'.
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"escalates an unusable last known revision" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      SomeException -> Maybe Text
retryableEtcdError (LastKnownRevisionException -> SomeException
forall e. Exception e => e -> SomeException
toException (LastKnownRevisionException -> SomeException)
-> LastKnownRevisionException -> SomeException
forall a b. (a -> b) -> a -> b
$ HostName -> LastKnownRevisionException
InvalidLastKnownRevision HostName
"peer/last-known-revision")
        Maybe Text -> Maybe Text -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Maybe Text
forall a. Maybe a
Nothing

  -- A file that holds no revision must not silently become revision 0: that
  -- rewinds the watch to the start of history. It takes the node down instead,
  -- so the error has to say how to recover.
  HostName -> Spec -> Spec
forall a. HasCallStack => HostName -> SpecWith a -> SpecWith a
describe HostName
"last known revision" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"round-trips a revision" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd-revision" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        HostName -> Natural -> IO ()
forall (m :: * -> *). MonadIO m => HostName -> Natural -> m ()
putLastKnownRevision HostName
tmp Natural
42
        HostName -> IO Natural
forall (m :: * -> *). MonadIO m => HostName -> m Natural
getLastKnownRevision HostName
tmp IO Natural -> Natural -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Natural
42

    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"starts from scratch when there is no file" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd-revision" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp ->
        HostName -> IO Natural
forall (m :: * -> *). MonadIO m => HostName -> m Natural
getLastKnownRevision HostName
tmp IO Natural -> Natural -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Natural
0

    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"rejects a file that holds no revision" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd-revision" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        let file :: HostName
file = HostName
tmp HostName -> HostName -> HostName
</> HostName
"last-known-revision"
        HostName -> ByteString -> IO ()
forall (m :: * -> *). MonadIO m => HostName -> ByteString -> m ()
writeFileBS HostName
file ByteString
""
        let isInvalid :: LastKnownRevisionException -> Bool
isInvalid = \case
              InvalidLastKnownRevision HostName
f -> HostName
f HostName -> HostName -> Bool
forall a. Eq a => a -> a -> Bool
== HostName
file
              UnreadableLastKnownRevision{} -> Bool
False
        HostName -> IO Natural
forall (m :: * -> *). MonadIO m => HostName -> m Natural
getLastKnownRevision HostName
tmp IO Natural -> (LastKnownRevisionException -> Bool) -> IO ()
forall e a.
(HasCallStack, Exception e) =>
IO a -> Selector e -> IO ()
`shouldThrow` LastKnownRevisionException -> Bool
isInvalid

    HostName -> IO () -> SpecM (Arg (IO ())) ()
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"says how to recover from a file that holds no revision" (IO () -> SpecM (Arg (IO ())) ())
-> IO () -> SpecM (Arg (IO ())) ()
forall a b. (a -> b) -> a -> b
$
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd-revision" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        let file :: HostName
file = HostName
tmp HostName -> HostName -> HostName
</> HostName
"last-known-revision"
        HostName -> ByteString -> IO ()
forall (m :: * -> *). MonadIO m => HostName -> ByteString -> m ()
writeFileBS HostName
file ByteString
"not a revision"
        IO Natural -> IO (Either LastKnownRevisionException 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 (HostName -> IO Natural
forall (m :: * -> *). MonadIO m => HostName -> m Natural
getLastKnownRevision HostName
tmp) IO (Either LastKnownRevisionException Natural)
-> (Either LastKnownRevisionException Natural -> 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
          Right (Natural
rev :: Natural) -> HostName -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
HostName -> m a
failure (HostName -> IO ()) -> HostName -> IO ()
forall a b. (a -> b) -> a -> b
$ HostName
"expected a failure, got revision " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> Natural -> HostName
forall b a. (Show a, IsString b) => a -> b
show Natural
rev
          Left (LastKnownRevisionException
e :: LastKnownRevisionException) ->
            LastKnownRevisionException -> HostName
forall e. Exception e => e -> HostName
displayException LastKnownRevisionException
e HostName -> HostName -> IO ()
forall a. (HasCallStack, Show a, Eq a) => [a] -> [a] -> IO ()
`shouldContain` (HostName
"Delete " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
file)

  HostName -> Spec -> Spec
forall a. HasCallStack => HostName -> SpecWith a -> SpecWith a
describe HostName
"Serialisation" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    HostName -> (Message SimpleTx -> Property) -> Spec
forall prop.
(HasCallStack, Testable prop) =>
HostName -> prop -> Spec
prop HostName
"can roundtrip CBOR encoding/decoding of Hydra Message" ((Message SimpleTx -> Property) -> Spec)
-> (Message SimpleTx -> Property) -> Spec
forall a b. (a -> b) -> a -> b
$ forall a. (ToCBOR a, FromCBOR a, Eq a, Show a) => a -> Property
prop_canRoundtripCBOREncoding @(Message SimpleTx)

    Settings -> Proxy (Message SimpleTx) -> Spec
forall a.
(Arbitrary a, ToADTArbitrary a, Eq a, Show a, ToJSON a,
 FromJSON a) =>
Settings -> Proxy a -> Spec
roundtripAndGoldenADTSpecsWithSettings Settings
defaultSettings{sampleSize = 1} (Proxy (Message SimpleTx) -> Spec)
-> Proxy (Message SimpleTx) -> Spec
forall a b. (a -> b) -> a -> b
$ forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @(Message SimpleTx)

-- | The etcd tests, kept out of 'spec' so that test/Main.hs can turn them
-- into a tasty group of their own and run them one at a time.
--
-- They must not run concurrently: each starts etcd on a port obtained by
-- binding a socket, reading the assigned port and closing it again, so two at
-- once can be handed the same port, or one test's peer port can be another's
-- derived client port. Whichever etcd binds second dies with EADDRINUSE, which
-- 'withEtcdNetwork' turns straight into a failure.
etcdSpec :: Spec
etcdSpec :: Spec
etcdSpec =
  ((Tracer IO EtcdLog -> IO ()) -> IO ())
-> SpecWith (Tracer IO EtcdLog) -> Spec
forall a. (ActionWith a -> IO ()) -> SpecWith a -> Spec
around (Text -> (Tracer IO EtcdLog -> IO ()) -> IO ()
forall (m :: * -> *) msg a.
(MonadLabelledSTM m, MonadCatch m, MonadFork m, MonadTime m,
 MonadSay m, ToJSON msg) =>
Text -> (Tracer m msg -> m a) -> m a
showLogsOnFailure Text
"NetworkSpec") (SpecWith (Tracer IO EtcdLog) -> Spec)
-> SpecWith (Tracer IO EtcdLog) -> Spec
forall a b. (a -> b) -> a -> b
$ do
    let v1 :: ProtocolVersion
v1 = Natural -> ProtocolVersion
ProtocolVersion Natural
1

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"broadcasts to self" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
30 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
          (PortNumber -> PortNumber) -> (PortNumber -> IO ()) -> IO ()
forall a.
(PortNumber -> PortNumber) -> (PortNumber -> IO a) -> IO a
withFreePortAndDerived PortNumber -> PortNumber
peerPortToClientPort ((PortNumber -> IO ()) -> IO ()) -> (PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \PortNumber
port -> do
            let config :: NetworkConfiguration
config =
                  NetworkConfiguration
                    { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port
                    , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port
                    , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
                    , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = []
                    , $sel:peers:NetworkConfiguration :: [Host]
peers = []
                    , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
                    , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
                    , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
                    }
            (NetworkCallback Text IO
recordingCallback, IO Text
waitNext, IO Connectivity
_) <- IO (NetworkCallback Text IO, IO Text, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
            Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO Text Text ()
forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
config NetworkCallback Text IO
recordingCallback ((Network IO Text -> IO ()) -> IO ())
-> (Network IO Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Text
n -> do
              Network IO Text -> Text -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Text
n (Text
"asdf" :: Text)
              IO Text
waitNext IO Text -> Text -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Text
"asdf"

    -- Exercises 'putMessage's compare-failure branch directly. The
    -- scenario it mimics: a previous 'putMessage' committed
    -- server-side but the gRPC client returned
    -- 'GrpcDeadlineExceeded', so the in-memory 'lastModRev' is
    -- stale on the retry. The Txn's compare(modRev == stale) must
    -- fail, the range branch must observe the new revision, the
    -- 'BroadcastDeduped' event must be traced, and most importantly
    -- /no second write/ should land — so the receiver does not see
    -- a duplicate.
    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"putMessage dedups when lastModRev is stale" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
30 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
          (PortNumber -> PortNumber) -> (PortNumber -> IO ()) -> IO ()
forall a.
(PortNumber -> PortNumber) -> (PortNumber -> IO a) -> IO a
withFreePortAndDerived PortNumber -> PortNumber
peerPortToClientPort ((PortNumber -> IO ()) -> IO ()) -> (PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \PortNumber
port -> do
            let host :: Host
host = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port
                config :: NetworkConfiguration
config =
                  NetworkConfiguration
                    { $sel:listen:NetworkConfiguration :: Host
listen = Host
host
                    , $sel:advertise:NetworkConfiguration :: Host
advertise = Host
host
                    , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
                    , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = []
                    , $sel:peers:NetworkConfiguration :: [Host]
peers = []
                    , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
                    , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
                    , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
                    }
            (NetworkCallback Int IO
recordingCallback, IO Int
waitNext, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
config NetworkCallback Int IO
recordingCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n -> do
              -- Real broadcast advances msg-<host>'s mod_revision.
              Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n Int
1
              IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
1
              -- Capture EtcdLog events from a direct putMessage call.
              TVar [Envelope EtcdLog]
traces <- HostName -> [Envelope EtcdLog] -> IO (TVar IO [Envelope EtcdLog])
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> a -> m (TVar m a)
newLabelledTVarIO HostName
"putMessage-dedup-traces" []
              let captureTracer :: Tracer IO EtcdLog
captureTracer = TVar IO [Envelope EtcdLog] -> Text -> Tracer IO EtcdLog
forall (m :: * -> *) msg.
(MonadFork m, MonadTime m, MonadSTM m) =>
TVar m [Envelope msg] -> Text -> Tracer m msg
traceInTVar TVar [Envelope EtcdLog]
TVar IO [Envelope EtcdLog]
traces Text
"putMessageDedupSpec"
              TVar Int64
staleVar <- HostName -> Int64 -> IO (TVar IO Int64)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> a -> m (TVar m a)
newLabelledTVarIO HostName
"stale-last-mod-rev" Int64
0
              -- Compare against 0 must fail (real modRev > 0); the
              -- failure branch should adopt the observed revision
              -- and trace BroadcastDeduped instead of writing.
              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
captureTracer 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 ->
                Tracer IO EtcdLog
-> Connection -> Host -> TVar IO Int64 -> ByteString -> IO ()
putMessage Tracer IO EtcdLog
captureTracer Connection
conn Host
host TVar Int64
TVar IO Int64
staleVar ([ByteString] -> ByteString
batchValue [Int -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' (Int
99 :: Int)])
              [EtcdLog]
captured <- (Envelope EtcdLog -> EtcdLog) -> [Envelope EtcdLog] -> [EtcdLog]
forall a b. (a -> b) -> [a] -> [b]
map Envelope EtcdLog -> EtcdLog
forall a. Envelope a -> a
message ([Envelope EtcdLog] -> [EtcdLog])
-> IO [Envelope EtcdLog] -> IO [EtcdLog]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar IO [Envelope EtcdLog] -> IO [Envelope EtcdLog]
forall a. TVar IO a -> IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar [Envelope EtcdLog]
TVar IO [Envelope EtcdLog]
traces
              [EtcdLog]
captured
                [EtcdLog] -> ([EtcdLog] -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (EtcdLog -> Bool) -> [EtcdLog] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any
                  ( \case
                      BroadcastDeduped{} -> Bool
True
                      EtcdLog
_ -> Bool
False
                  )
              Int64
updatedModRev <- TVar IO Int64 -> IO Int64
forall a. TVar IO a -> IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar Int64
TVar IO Int64
staleVar
              Int64
updatedModRev Int64 -> (Int64 -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
> Int64
0)
              -- Send a fresh marker. If 99 had actually been
              -- written by the deduped call, the receiver would
              -- see it before 2 (etcd revisions are monotonic and
              -- the watch is in-order).
              Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n Int
2
              IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
2

    -- Regression test for #2817. etcd's grpc-go server raises its receive
    -- window as inbound volume grows, emitting a SETTINGS frame per step, and
    -- 'http2' rate limits inbound non-ACK SETTINGS to 4/s per connection
    -- (CVE-2019-9515), killing the connection on the 5th. Under real load the
    -- ramp is walked gradually as the broadcast queue backs up, which is why
    -- it only bit some runs; here the burst is injected rather than provoked,
    -- see 'withSettingsBurstProxy'.
    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"survives a burst of SETTINGS frames from etcd" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
          (PortNumber -> PortNumber) -> (PortNumber -> IO ()) -> IO ()
forall a.
(PortNumber -> PortNumber) -> (PortNumber -> IO a) -> IO a
withFreePortAndDerived PortNumber -> PortNumber
peerPortToClientPort ((PortNumber -> IO ()) -> IO ()) -> (PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \PortNumber
port -> do
            let host :: Host
host = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port
                config :: NetworkConfiguration
config =
                  NetworkConfiguration
                    { $sel:listen:NetworkConfiguration :: Host
listen = Host
host
                    , $sel:advertise:NetworkConfiguration :: Host
advertise = Host
host
                    , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
                    , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = []
                    , $sel:peers:NetworkConfiguration :: [Host]
peers = []
                    , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
                    , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
                    , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
                    }
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Text Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
config NetworkCallback Text IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Text -> IO ()) -> IO ())
-> (Network IO Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Text
_ ->
              -- One more than http2's limit of 4/s.
              PortNumber -> Int -> (PortNumber -> IO Int -> IO ()) -> IO ()
forall a.
PortNumber -> Int -> (PortNumber -> IO Int -> IO a) -> IO a
withSettingsBurstProxy (NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration
config) Int
5 ((PortNumber -> IO Int -> IO ()) -> IO ())
-> (PortNumber -> IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \PortNumber
proxyPort IO Int
settingsSeen -> do
                -- 'withEtcdNetwork' returns before etcd accepts clients, so
                -- gate on a direct query, whose reconnect policy waits.
                -- Doubles as the 'lastModRev' seed.
                TVar Int64
lastModRevVar <-
                  HostName -> Int64 -> IO (TVar IO Int64)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> a -> m (TVar m a)
newLabelledTVarIO HostName
"settings-burst-last-mod-rev"
                    (Int64 -> IO (TVar Int64)) -> IO Int64 -> IO (TVar Int64)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< Tracer IO EtcdLog -> NetworkConfiguration -> Host -> IO Int64
queryInitialModRev Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
host
                let proxied :: Server
proxied =
                      Address -> Server
ServerInsecure
                        Address
                          { addressHost :: HostName
addressHost = HostName
forall s. IsString s => s
lo
                          , addressPort :: PortNumber
addressPort = PortNumber
proxyPort
                          , addressAuthority :: Maybe HostName
addressAuthority = Maybe HostName
forall a. Maybe a
Nothing
                          }
                    -- Big enough that the burst lands while the put is still
                    -- streaming, which is when it arrives in production.
                    value :: ByteString
value = [ByteString] -> ByteString
batchValue [Text -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Int -> Text -> Text
T.replicate (Int
512 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1024) Text
"a"]
                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) Server
proxied ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
                  Tracer IO EtcdLog
-> Connection -> Host -> TVar IO Int64 -> ByteString -> IO ()
putMessage Tracer IO EtcdLog
tracer Connection
conn Host
host TVar Int64
TVar IO Int64
lastModRevVar ByteString
value
                -- Not a vacuous pass: 5 injected plus etcd's own handshake
                -- frame, and possibly more from its window ramp.
                IO Int
settingsSeen IO Int -> (Int -> 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
>>= (Int -> (Int -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
6))

    -- Note: This test is disabled as it takes took long; but it is
    -- important to keep around. Successfully completion of this test looks
    -- like either a "mvcc database size exceeded" error; or no error at
    -- all. Failures looks like complete blocking
    (IO () -> IO ())
-> SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog)
forall a. (IO () -> IO ()) -> SpecWith a -> SpecWith a
around_ IO () -> IO ()
onlyLocal (SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog))
-> SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog)
forall a b. (a -> b) -> a -> b
$ HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
xit HostName
"broadcasts 100KiB messages 1M times" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer ->
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        HostName -> IO ()
forall (m :: * -> *). MonadIO m => HostName -> m ()
putStrLn (HostName -> IO ()) -> HostName -> IO ()
forall a b. (a -> b) -> a -> b
$ HostName
"Folder " HostName -> HostName -> HostName
forall a. [a] -> [a] -> [a]
++ HostName -> HostName
forall b a. (Show a, IsString b) => a -> b
show HostName
tmp
        PeerConfig2{NetworkConfiguration
aliceConfig :: NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig, NetworkConfiguration
bobConfig :: NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
        (NetworkCallback Text IO
recordReceived, IO Text
waitNext, IO Connectivity
_) <- IO (NetworkCallback Text IO, IO Text, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
        -- Create a 100KiB message (100 * 1024 characters)
        let largeMessage :: Text
largeMessage = HostName -> Text
forall a. ToText a => a -> Text
toText (HostName -> Text) -> HostName -> Text
forall a b. (a -> b) -> a -> b
$ Int -> Char -> HostName
forall a. Int -> a -> [a]
replicate (Int
100 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1024) Char
'a'
        forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Text Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Text IO
recordReceived ((Network IO Text -> IO ()) -> IO ())
-> (Network IO Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Text
n1 -> do
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Text Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Text IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Text -> IO ()) -> IO ())
-> (Network IO Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Text
_ -> do
            [Integer] -> (Integer -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Integer
1 :: Integer .. Integer
1000000] ((Integer -> IO ()) -> IO ()) -> (Integer -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Integer
i -> do
              let msgWithId :: Text
msgWithId = Text
largeMessage Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" - Message #" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Integer -> Text
forall b a. (Show a, IsString b) => a -> b
show Integer
i
              Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Integer
i Integer -> Integer -> Integer
forall a. Integral a => a -> a -> a
`mod` Integer
10000 Integer -> Integer -> Bool
forall a. Eq a => a -> a -> Bool
== Integer
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                HostName -> IO ()
forall (m :: * -> *). MonadIO m => HostName -> m ()
putStrLn (HostName -> IO ()) -> HostName -> IO ()
forall a b. (a -> b) -> a -> b
$
                  HostName
"Broadcasting 100KiB message #" HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> Integer -> HostName
forall b a. (Show a, IsString b) => a -> b
show Integer
i HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
" (size: " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> Int -> HostName
forall b a. (Show a, IsString b) => a -> b
show (HostName -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length (Text -> HostName
forall a. ToString a => a -> HostName
toString Text
msgWithId)) HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
" chars)"
              Network IO Text -> Text -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Text
n1 Text
msgWithId
              Text
_ <- IO Text
waitNext
              DiffTime -> IO ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
0.02

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"broadcasts messages to single connected peer" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
30 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 -> do
            (NetworkCallback Int IO
recordReceived, IO Int
waitNext, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_n2 -> do
              Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
123
              IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
123

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"handles broadcast to minority" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig3{NetworkConfiguration
aliceConfig :: NetworkConfiguration
$sel:aliceConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
aliceConfig, NetworkConfiguration
bobConfig :: NetworkConfiguration
$sel:bobConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
bobConfig, NetworkConfiguration
carolConfig :: NetworkConfiguration
$sel:carolConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
carolConfig} <- HostName -> IO PeerConfig3
setup3Peers HostName
tmp
          (NetworkCallback Int IO
recordReceived, IO Int
waitNext, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 -> do
            -- Bob and carol start and stop
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            -- Alice sends a message while she is the only one online (= minority)
            Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
123
          -- Now, alice stops too!
          -- Start alice, bob and carol again
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                -- Alice should see her own message eventually (when part of majority again)
                IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
123

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"handles broadcast to majority" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig3{NetworkConfiguration
$sel:aliceConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig, NetworkConfiguration
$sel:carolConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
carolConfig :: NetworkConfiguration
carolConfig} <- HostName -> IO PeerConfig3
setup3Peers HostName
tmp
          (NetworkCallback Int IO
recordReceived, IO Int
waitNext, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                -- Alice sends a message while Carol is online
                Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
123
                IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
123
              -- Alice sends a message while Carol is offline
              Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
456
              -- Carol starts again
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                -- Carol should receive messages sent by alice while offline
                -- (without duplication of 123)
                IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
456

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"emits connectivity events" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig3{NetworkConfiguration
$sel:aliceConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig, NetworkConfiguration
$sel:carolConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
carolConfig :: NetworkConfiguration
carolConfig} <- HostName -> IO PeerConfig3
setup3Peers HostName
tmp
          -- Record and assert connectivity events from alice's perspective
          (NetworkCallback Int IO
recordReceived, IO Int
_, IO Connectivity
waitConnectivity) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          let
            waitFor :: HasCallStack => Connectivity -> IO ()
            waitFor :: HasCallStack => Connectivity -> IO ()
waitFor = IO Connectivity -> NominalDiffTime -> Connectivity -> IO ()
forall a.
(HasCallStack, Eq a, Show a) =>
IO a -> NominalDiffTime -> a -> IO ()
waitEq IO Connectivity
waitConnectivity NominalDiffTime
60
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              -- Alice now on majority cluster
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor Connectivity
NetworkConnected
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerConnected NetworkConfiguration
bobConfig.advertise
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerConnected NetworkConfiguration
carolConfig.advertise
                -- Carol stops
                () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerDisconnected NetworkConfiguration
carolConfig.advertise
              -- Bob stops
              () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            -- We are now in minority
            HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor Connectivity
NetworkDisconnected
            -- Carol starts again and we reach a majority
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor Connectivity
NetworkConnected
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerConnected NetworkConfiguration
carolConfig.advertise

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"handles expired lease" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
30 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          -- Record and assert connectivity events from alice's perspective
          (NetworkCallback Int IO
recordReceived, IO Int
_, IO Connectivity
waitConnectivity) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          let
            waitFor :: HasCallStack => Connectivity -> IO ()
            waitFor :: HasCallStack => Connectivity -> IO ()
waitFor = IO Connectivity -> NominalDiffTime -> Connectivity -> IO ()
forall a.
(HasCallStack, Eq a, Show a) =>
IO a -> NominalDiffTime -> a -> IO ()
waitEq IO Connectivity
waitConnectivity NominalDiffTime
60
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordReceived ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor Connectivity
NetworkConnected
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerConnected NetworkConfiguration
bobConfig.advertise
              -- Expire all leases manually to simulate a keepAlive coming too
              -- late. Note that we do not distinguish which is which so
              -- alice's lease will also be killed, but does not matter here.
              -- NOTE: etcdctl talks to the etcd /client/ port, not the peer
              -- port. Using @listen aliceConfig@ here used to work on etcd
              -- 3.5 (whose peer listener happened to answer client RPCs too)
              -- but hangs on etcd 3.6, where the peer port no longer serves
              -- the lease API.
              let endpoints :: HostName
endpoints = HostName
"--endpoints=127.0.0.1:" HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> PortNumber -> HostName
forall b a. (Show a, IsString b) => a -> b
show (NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration
aliceConfig)
              ByteString
output <- ProcessConfig () () () -> IO ByteString
forall (m :: * -> *) stdin stdoutIgnored stderr.
MonadIO m =>
ProcessConfig stdin stdoutIgnored stderr -> m ByteString
readProcessStdout_ (ProcessConfig () () () -> IO ByteString)
-> (HostName -> ProcessConfig () () ())
-> HostName
-> IO ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HostName -> ProcessConfig () () ()
shell (HostName -> IO ByteString) -> HostName -> IO ByteString
forall a b. (a -> b) -> a -> b
$ HostName
"etcdctl lease list " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
endpoints
              let leases :: [Text]
leases = Int -> [Text] -> [Text]
forall a. Int -> [a] -> [a]
drop Int
1 ([Text] -> [Text]) -> [Text] -> [Text]
forall a b. (a -> b) -> a -> b
$ Text -> [Text]
forall t. IsText t "lines" => t -> [t]
lines (Text -> [Text]) -> Text -> [Text]
forall a b. (a -> b) -> a -> b
$ ByteString -> Text
forall a b. ConvertUtf8 a b => b -> a
decodeUtf8 ByteString
output
              [Text] -> (Text -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Text]
leases ((Text -> IO ()) -> IO ()) -> (Text -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Text
lease ->
                ProcessConfig () () () -> IO ()
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
ProcessConfig stdin stdout stderr -> m ()
runProcess_ (ProcessConfig () () () -> IO ())
-> (HostName -> ProcessConfig () () ()) -> HostName -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HostName -> ProcessConfig () () ()
shell (HostName -> IO ()) -> HostName -> IO ()
forall a b. (a -> b) -> a -> b
$ HostName
"etcdctl lease revoke " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
endpoints HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> HostName
" " HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> Text -> HostName
forall a. ToString a => a -> HostName
toString Text
lease
              -- Alice sees bob disconnected and connected again
              HasCallStack => Connectivity -> IO ()
Connectivity -> IO ()
waitFor (Connectivity -> IO ()) -> Connectivity -> IO ()
forall a b. (a -> b) -> a -> b
$ Host -> Connectivity
PeerConnected NetworkConfiguration
bobConfig.advertise

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"checks protocol version" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          let v2 :: ProtocolVersion
v2 = Natural -> ProtocolVersion
ProtocolVersion Natural
2
          (NetworkCallback Int IO
recordAlice, IO Int
_, IO Connectivity
waitAlice) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          (NetworkCallback Int IO
recordBob, IO Int
_, IO Connectivity
waitBob) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          let aliceSees :: Connectivity -> IO ()
aliceSees = IO Connectivity -> NominalDiffTime -> Connectivity -> IO ()
forall a.
(HasCallStack, Eq a, Show a) =>
IO a -> NominalDiffTime -> a -> IO ()
waitEq IO Connectivity
waitAlice NominalDiffTime
30
              bobSees :: Connectivity -> IO ()
bobSees = IO Connectivity -> NominalDiffTime -> Connectivity -> IO ()
forall a.
(HasCallStack, Eq a, Show a) =>
IO a -> NominalDiffTime -> a -> IO ()
waitEq IO Connectivity
waitBob NominalDiffTime
30
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordAlice ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v2 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              -- Both will try to write to the cluster at the same time
              -- Hence, either one or the other will see the mismatch
              (HostName, IO ()) -> (HostName, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(HostName, m a) -> (HostName, m b) -> m ()
raceLabelled_
                (HostName
"alice-sees", Connectivity -> IO ()
aliceSees VersionMismatch{$sel:ourVersion:PeerConnected :: ProtocolVersion
ourVersion = ProtocolVersion
v1, $sel:theirVersion:PeerConnected :: Maybe ProtocolVersion
theirVersion = ProtocolVersion -> Maybe ProtocolVersion
forall a. a -> Maybe a
Just ProtocolVersion
v2})
                (HostName
"bob-sees", Connectivity -> IO ()
bobSees VersionMismatch{$sel:ourVersion:PeerConnected :: ProtocolVersion
ourVersion = ProtocolVersion
v2, $sel:theirVersion:PeerConnected :: Maybe ProtocolVersion
theirVersion = ProtocolVersion -> Maybe ProtocolVersion
forall a. a -> Maybe a
Just ProtocolVersion
v1})

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"resends messages" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        -- Sends 1000 messages through a 3-node etcd cluster; the
        -- 20s budget was too tight under parallel CI load.
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig3{NetworkConfiguration
$sel:aliceConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig, NetworkConfiguration
$sel:carolConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
carolConfig :: NetworkConfiguration
carolConfig} <- HostName -> IO PeerConfig3
setup3Peers HostName
tmp
          (NetworkCallback Int IO
recordBob, IO Int
waitBob, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          (NetworkCallback Int IO
recordCarol, IO Int
waitCarol, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              let messages :: [Int]
messages = [Int
1 .. Int
1000]
              -- Bob should see messages as we go
              [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int]
messages ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
msg
                IO Int
waitBob IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
              -- Carol only starts now and should see all messages delivered
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordCarol ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int]
messages ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg ->
                  IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
              -- Carol only delivers new messages even after restart
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordCarol ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
1001
                IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
1001

    -- Sequential broadcasts force one etcd put each (batching only kicks
    -- in when the queue backs up), so 5000 sends cross the broadcast
    -- connection recycle boundary several times. Guards against the
    -- long-lived connection blocking (issue #2167) with the reused
    -- connection. Takes a few minutes, hence local-only.
    (IO () -> IO ())
-> SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog)
forall a. (IO () -> IO ()) -> SpecWith a -> SpecWith a
around_ IO () -> IO ()
onlyLocal (SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog))
-> SpecWith (Tracer IO EtcdLog) -> SpecWith (Tracer IO EtcdLog)
forall a b. (a -> b) -> a -> b
$ HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"sustains sequential broadcasts across connection recycles" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer ->
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
600 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          (NetworkCallback Int IO
recordBob, IO Int
waitBob, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ ->
              [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
1 .. Int
5000 :: Int] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
msg
                IO Int
waitBob IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"batches queued messages and delivers them in order" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          (NetworkCallback Int IO
recordBob, IO Int
waitBob, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              -- Broadcast without waiting in between, so the sender's queue
              -- accumulates and messages travel as multi-message batch
              -- values. Delivery must still be in order, exactly once.
              [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
1 .. Int
200] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1
              [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
1 .. Int
200] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> IO Int
waitBob IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"delivers legacy single-message values" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
30 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
          (PortNumber -> PortNumber) -> (PortNumber -> IO ()) -> IO ()
forall a.
(PortNumber -> PortNumber) -> (PortNumber -> IO a) -> IO a
withFreePortAndDerived PortNumber -> PortNumber
peerPortToClientPort ((PortNumber -> IO ()) -> IO ()) -> (PortNumber -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \PortNumber
port -> do
            let host :: Host
host = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port
                config :: NetworkConfiguration
config =
                  NetworkConfiguration
                    { $sel:listen:NetworkConfiguration :: Host
listen = Host
host
                    , $sel:advertise:NetworkConfiguration :: Host
advertise = Host
host
                    , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
                    , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = []
                    , $sel:peers:NetworkConfiguration :: [Host]
peers = []
                    , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
                    , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
                    , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
                    }
            (NetworkCallback Int IO
recordingCallback, IO Int
waitNext, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
config NetworkCallback Int IO
recordingCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n -> do
              Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n Int
1
              IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
1
              -- Write a value in the pre-batching wire format (a single
              -- CBOR message, not a list); the watch must deliver it
              -- through the legacy fallback decoder.
              Int64
rev <- Tracer IO EtcdLog -> NetworkConfiguration -> Host -> IO Int64
queryInitialModRev Tracer IO EtcdLog
tracer NetworkConfiguration
config Host
host
              TVar Int64
revVar <- HostName -> Int64 -> IO (TVar IO Int64)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> a -> m (TVar m a)
newLabelledTVarIO HostName
"legacy-value-mod-rev" Int64
rev
              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 ->
                Tracer IO EtcdLog
-> Connection -> Host -> TVar IO Int64 -> ByteString -> IO ()
putMessage Tracer IO EtcdLog
tracer Connection
conn Host
host TVar Int64
TVar IO Int64
revVar (Int -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' (Int
7 :: Int))
              IO Int
waitNext IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
7

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"handles compaction and lost local state" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig3{NetworkConfiguration
$sel:aliceConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig, NetworkConfiguration
$sel:carolConfig:PeerConfig3 :: PeerConfig3 -> NetworkConfiguration
carolConfig :: NetworkConfiguration
carolConfig} <- HostName -> IO PeerConfig3
setup3Peers HostName
tmp
          (NetworkCallback Int IO
recordBob, IO Int
waitBob, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          (NetworkCallback Int IO
recordCarol, IO Int
waitCarol, IO Connectivity
_) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
forall (m :: * -> *) msg. Applicative m => NetworkCallback msg m
noopCallback ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
n1 ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
bobConfig NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
              -- First we send 5 messages with carol online
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordCarol ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
1 .. Int
5] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                  Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
msg
                  IO Int
waitBob IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
                  IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
              -- Carol stopped and we continue sending messages
              [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
5 .. Int
100] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
msg
                IO Int
waitBob IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
              -- Even while carol is down, the etcd component would
              -- "auto-compact" messages. By default down to 1000 messages
              -- after/every 5 minutes. This is interesting as it should
              -- result in carol never some messages, but is hard to test
              -- (without waiting 5 minutes). Instead we issue a direct etcd
              -- command to compact everything before revision 50.
              ProcessConfig () () () -> IO ()
forall (m :: * -> *) stdin stdout stderr.
MonadIO m =>
ProcessConfig stdin stdout stderr -> m ()
runProcess_ (ProcessConfig () () () -> IO ())
-> (HostName -> ProcessConfig () () ()) -> HostName -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HostName -> ProcessConfig () () ()
shell (HostName -> IO ()) -> HostName -> IO ()
forall a b. (a -> b) -> a -> b
$
                HostName
"etcdctl compact 50 --endpoints=127.0.0.1:" HostName -> HostName -> HostName
forall a. Semigroup a => a -> a -> a
<> PortNumber -> HostName
forall b a. (Show a, IsString b) => a -> b
show (NetworkConfiguration -> PortNumber
getClientPort NetworkConfiguration
aliceConfig)
              -- When carol starts now we would expect it to start catching up
              -- from the earliest possible revision 50. While missing some
              -- messages.
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordCarol ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                -- NOTE: Revision 50 may not correspond to message 50, so we
                -- only assert its some message bigger than 25 and expect to
                -- see all further messages to 100.
                Int
firstMsg <- IO Int
waitCarol
                Int
firstMsg Int -> (Int -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
25)
                [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
firstMsg Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1 .. Int
100] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg ->
                  IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
                -- Carol should be able to receive new messages just fine.
                [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
101 .. Int
105] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                  Network IO Int -> Int -> IO ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network IO Int
n1 Int
msg
                  IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg
              -- Similarly, should carol lose its local state, we expect it to
              -- see everything from the last compacted revision 50. We can
              -- enforce this by removing the corresponding file (an internal
              -- implementation detail)
              HostName -> IO ()
removeFile (NetworkConfiguration -> HostName
persistenceDir NetworkConfiguration
carolConfig HostName -> HostName -> HostName
</> HostName
"last-known-revision")
              forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
carolConfig NetworkCallback Int IO
recordCarol ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ -> do
                -- NOTE: Revision 50 may not correspond to message 50, so we
                -- only assert its some message bigger than 25 and expect to
                -- see all further messages to 105.
                Int
firstMsg <- IO Int
waitCarol
                Int
firstMsg Int -> (Int -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` (Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
25)
                [Int] -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Int
firstMsg Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1 .. Int
105] ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
msg -> do
                  IO Int
waitCarol IO Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => IO a -> a -> IO ()
`shouldReturn` Int
msg

    HostName
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a.
(HasCallStack, Example a) =>
HostName -> a -> SpecWith (Arg a)
it HostName
"emits cluster id mismatch" ((Tracer IO EtcdLog -> IO ())
 -> SpecWith (Arg (Tracer IO EtcdLog -> IO ())))
-> (Tracer IO EtcdLog -> IO ())
-> SpecWith (Arg (Tracer IO EtcdLog -> IO ()))
forall a b. (a -> b) -> a -> b
$ \Tracer IO EtcdLog
tracer -> do
      HostName -> (HostName -> IO ()) -> IO ()
forall (m :: * -> *) r.
(MonadIO m, MonadMask m) =>
HostName -> (HostName -> m r) -> m r
withTempDir HostName
"test-etcd" ((HostName -> IO ()) -> IO ()) -> (HostName -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \HostName
tmp -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
60 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          PeerConfig2{NetworkConfiguration
$sel:aliceConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
aliceConfig, NetworkConfiguration
$sel:bobConfig:PeerConfig2 :: PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
bobConfig} <- HostName -> IO PeerConfig2
setup2Peers HostName
tmp
          let v2 :: ProtocolVersion
v2 = Natural -> ProtocolVersion
ProtocolVersion Natural
2
          (NetworkCallback Int IO
recordAlice, IO Int
_, IO Connectivity
waitAlice) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          (NetworkCallback Int IO
recordBob, IO Int
_, IO Connectivity
waitBob) <- IO (NetworkCallback Int IO, IO Int, IO Connectivity)
forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback
          let aliceSees :: (Connectivity -> Maybe ()) -> IO ()
aliceSees = IO Connectivity
-> NominalDiffTime -> (Connectivity -> Maybe ()) -> IO ()
forall a b.
(HasCallStack, Show a) =>
IO a -> NominalDiffTime -> (a -> Maybe b) -> IO b
waitMatch IO Connectivity
waitAlice NominalDiffTime
30
          let bobSees :: (Connectivity -> Maybe ()) -> IO ()
bobSees = IO Connectivity
-> NominalDiffTime -> (Connectivity -> Maybe ()) -> IO ()
forall a b.
(HasCallStack, Show a) =>
IO a -> NominalDiffTime -> (a -> Maybe b) -> IO b
waitMatch IO Connectivity
waitBob NominalDiffTime
30
          let bobConfig' :: NetworkConfiguration
bobConfig' = NetworkConfiguration
bobConfig{peers = []}
          forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v1 NetworkConfiguration
aliceConfig NetworkCallback Int IO
recordAlice ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ ->
            forall msg.
(ToCBOR msg, FromCBOR msg) =>
Tracer IO EtcdLog
-> ProtocolVersion
-> NetworkConfiguration
-> NetworkComponent IO msg msg ()
withEtcdNetwork @Int Tracer IO EtcdLog
tracer ProtocolVersion
v2 NetworkConfiguration
bobConfig' NetworkCallback Int IO
recordBob ((Network IO Int -> IO ()) -> IO ())
-> (Network IO Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Network IO Int
_ ->
              (HostName, IO ()) -> (HostName, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(HostName, m a) -> (HostName, m b) -> m ()
raceLabelled_
                (HostName
"bob-sees", (Connectivity -> Maybe ()) -> IO ()
bobSees ((Connectivity -> Maybe ()) -> IO ())
-> (Connectivity -> Maybe ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \case ClusterIDMismatch{} -> () -> Maybe ()
forall a. a -> Maybe a
Just (); Connectivity
_ -> Maybe ()
forall a. Maybe a
Nothing)
                (HostName
"alice-sees", (Connectivity -> Maybe ()) -> IO ()
aliceSees ((Connectivity -> Maybe ()) -> IO ())
-> (Connectivity -> Maybe ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \case ClusterIDMismatch{} -> () -> Maybe ()
forall a. a -> Maybe a
Just (); Connectivity
_ -> Maybe ()
forall a. Maybe a
Nothing)

lo :: IsString s => s
lo :: forall s. IsString s => s
lo = s
"127.0.0.1"

-- | Run a TCP proxy in front of an etcd client port that reproduces #2817's
-- SETTINGS burst deterministically.
--
-- Both directions are forwarded verbatim, and once the client starts sending
-- request data (so an RPC is in flight, as in production, where the data being
-- sent is what makes etcd's window ramp fire) the proxy pushes @burst@ empty
-- SETTINGS frames at the client. Those are legal at any point; the client ACKs
-- each and etcd ignores stray ACKs.
--
-- Injecting beats provoking: how many frames grpc-go's estimator emits depends
-- on how much data lands in one round-trip, which on an idle machine stops at
-- 3, one under the limit. Reaching 5 needs the scheduling latency of the
-- reporter's six-nodes-on-one-host setup.
withSettingsBurstProxy ::
  -- | Upstream etcd client port.
  PortNumber ->
  -- | How many SETTINGS frames to burst at the client.
  Int ->
  -- | Given the proxy's port and the count of non-ACK SETTINGS frames
  -- forwarded to the client so far.
  (PortNumber -> IO Int -> IO a) ->
  IO a
withSettingsBurstProxy :: forall a.
PortNumber -> Int -> (PortNumber -> IO Int -> IO a) -> IO a
withSettingsBurstProxy PortNumber
upstreamPort Int
burst PortNumber -> IO Int -> IO a
action = do
  TVar Int
settingsSeen <- HostName -> Int -> IO (TVar IO Int)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> a -> m (TVar m a)
newLabelledTVarIO HostName
"settings-burst-seen" Int
0
  IO Socket -> (Socket -> IO ()) -> (Socket -> IO a) -> IO a
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket IO Socket
listenLoopback Socket -> IO ()
close ((Socket -> IO a) -> IO a) -> (Socket -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Socket
server -> do
    PortNumber
proxyPort <- Socket -> IO PortNumber
socketPort Socket
server
    (HostName, IO Any) -> (Async IO Any -> IO a) -> IO a
forall (m :: * -> *) a b.
MonadAsync m =>
(HostName, m a) -> (Async m a -> m b) -> m b
withAsyncLabelled (HostName
"settings-burst-proxy", TVar Int -> Socket -> IO Any
acceptLoop TVar Int
settingsSeen Socket
server) ((Async IO Any -> IO a) -> IO a) -> (Async IO Any -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async IO Any
_ ->
      PortNumber -> IO Int -> IO a
action PortNumber
proxyPort (TVar IO Int -> IO Int
forall a. TVar IO a -> IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar Int
TVar IO Int
settingsSeen)
 where
  loopback :: HostAddress
loopback = (Word8, Word8, Word8, Word8) -> HostAddress
tupleToHostAddress (Word8
127, Word8
0, Word8
0, Word8
1)

  listenLoopback :: IO Socket
listenLoopback = do
    Socket
sock <- Family -> SocketType -> ProtocolNumber -> IO Socket
socket Family
AF_INET SocketType
Stream ProtocolNumber
defaultProtocol
    Socket -> SocketOption -> Int -> IO ()
setSocketOption Socket
sock SocketOption
ReuseAddr Int
1
    Socket -> SockAddr -> IO ()
bind Socket
sock (SockAddr -> IO ()) -> SockAddr -> IO ()
forall a b. (a -> b) -> a -> b
$ PortNumber -> HostAddress -> SockAddr
SockAddrInet PortNumber
0 HostAddress
loopback
    Socket -> Int -> IO ()
Socket.listen Socket
sock Int
5
    Socket -> IO Socket
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Socket
sock

  acceptLoop :: TVar Int -> Socket -> IO Any
acceptLoop TVar Int
settingsSeen Socket
server = IO () -> IO Any
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO Any) -> IO () -> IO Any
forall a b. (a -> b) -> a -> b
$ do
    (Socket
client, SockAddr
_) <- Socket -> IO (Socket, SockAddr)
accept Socket
server
    -- Socket teardown at the end of a test races the relay threads; a dead
    -- proxy connection has nothing left to report, so let it go quietly.
    IO (Async ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Async ()) -> IO ())
-> (IO () -> IO (Async ())) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HostName -> IO () -> IO (Async IO ())
forall (m :: * -> *) a.
MonadAsync m =>
HostName -> m a -> m (Async m a)
asyncLabelled HostName
"settings-burst-proxy-connection" (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      (IO (Either SomeException ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO (Either SomeException ()) -> IO ())
-> (IO () -> IO (Either SomeException ())) -> IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try @_ @SomeException (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar Int -> Socket -> IO ()
relay TVar Int
settingsSeen Socket
client) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
forall (m :: * -> *) a b. MonadThrow m => m a -> m b -> m a
`finally` Socket -> IO ()
close Socket
client

  relay :: TVar Int -> Socket -> IO ()
relay TVar Int
settingsSeen Socket
client =
    IO Socket -> (Socket -> IO ()) -> (Socket -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket IO Socket
connectUpstream Socket -> IO ()
close ((Socket -> IO ()) -> IO ()) -> (Socket -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Socket
upstream -> do
      -- Single writer towards the client, so forwarded bytes and the injected
      -- burst cannot interleave mid-frame.
      TBQueue ByteString
toClient <- HostName -> Natural -> IO (TBQueue IO ByteString)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> Natural -> m (TBQueue m a)
newLabelledTBQueueIO HostName
"settings-burst-to-client" Natural
100
      (HostName, IO Any) -> (HostName, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(HostName, m a) -> (HostName, m b) -> m ()
raceLabelled_
        (HostName
"settings-burst-writer", TVar Int -> TBQueue ByteString -> Socket -> ByteString -> IO Any
writeToClient TVar Int
settingsSeen TBQueue ByteString
toClient Socket
client ByteString
"")
        ( HostName
"settings-burst-readers"
        , (HostName, IO ()) -> (HostName, IO ()) -> IO ()
forall (m :: * -> *) a b.
MonadAsync m =>
(HostName, m a) -> (HostName, m b) -> m ()
raceLabelled_
            (HostName
"settings-burst-upstream-reader", Socket -> (ByteString -> IO ()) -> IO ()
readInto Socket
upstream (STM () -> IO ()
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 ()) -> (ByteString -> STM ()) -> ByteString -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TBQueue IO ByteString -> ByteString -> STM IO ()
forall a. TBQueue IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue ByteString
TBQueue IO ByteString
toClient))
            (HostName
"settings-burst-client-reader", TBQueue ByteString -> Socket -> Socket -> Int -> IO ()
readFromClient TBQueue ByteString
toClient Socket
client Socket
upstream Int
0)
        )

  connectUpstream :: IO Socket
connectUpstream = do
    Socket
sock <- Family -> SocketType -> ProtocolNumber -> IO Socket
socket Family
AF_INET SocketType
Stream ProtocolNumber
defaultProtocol
    Socket -> SockAddr -> IO ()
connect Socket
sock (SockAddr -> IO ()) -> SockAddr -> IO ()
forall a b. (a -> b) -> a -> b
$ PortNumber -> HostAddress -> SockAddr
SockAddrInet PortNumber
upstreamPort HostAddress
loopback
    Socket -> IO Socket
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Socket
sock

  readInto :: Socket -> (ByteString -> IO ()) -> IO ()
readInto Socket
from ByteString -> IO ()
sink = do
    ByteString
bs <- Socket -> Int -> IO ByteString
recv Socket
from Int
chunkSize
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (ByteString -> Bool
BS.null ByteString
bs) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      ByteString -> IO ()
sink ByteString
bs
      Socket -> (ByteString -> IO ()) -> IO ()
readInto Socket
from ByteString -> IO ()
sink

  writeToClient :: TVar Int -> TBQueue ByteString -> Socket -> ByteString -> IO Any
writeToClient TVar Int
settingsSeen TBQueue ByteString
toClient Socket
client ByteString
buffer = do
    ByteString
bs <- STM IO ByteString -> IO ByteString
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO ByteString -> IO ByteString)
-> STM IO ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$ TBQueue IO ByteString -> STM IO ByteString
forall a. TBQueue IO a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m a
readTBQueue TBQueue ByteString
TBQueue IO ByteString
toClient
    Socket -> ByteString -> IO ()
sendAll Socket
client ByteString
bs
    ByteString
buffer' <- TVar Int -> ByteString -> IO ByteString
countSettingsFrames TVar Int
settingsSeen (ByteString
buffer ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
bs)
    TVar Int -> TBQueue ByteString -> Socket -> ByteString -> IO Any
writeToClient TVar Int
settingsSeen TBQueue ByteString
toClient Socket
client ByteString
buffer'

  -- Trigger off bytes sent rather than parsing client frames: anything past the
  -- 24-byte preface and the SETTINGS + HEADERS frames is request body.
  readFromClient :: TBQueue ByteString -> Socket -> Socket -> Int -> IO ()
readFromClient TBQueue ByteString
toClient Socket
client Socket
upstream Int
sent = do
    ByteString
bs <- Socket -> Int -> IO ByteString
recv Socket
client Int
chunkSize
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (ByteString -> Bool
BS.null ByteString
bs) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      Socket -> ByteString -> IO ()
sendAll Socket
upstream ByteString
bs
      let sent' :: Int
sent' = Int
sent Int -> Int -> Int
forall a. Num a => a -> a -> a
+ ByteString -> Int
BS.length ByteString
bs
      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
sent Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
burstAfterBytes Bool -> Bool -> Bool
&& Int
sent' Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
burstAfterBytes) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        STM () -> IO ()
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 ())
-> ([ByteString] -> STM ()) -> [ByteString] -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TBQueue IO ByteString -> ByteString -> STM IO ()
forall a. TBQueue IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue ByteString
TBQueue IO ByteString
toClient (ByteString -> STM ())
-> ([ByteString] -> ByteString) -> [ByteString] -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [ByteString] -> ByteString
BS.concat ([ByteString] -> IO ()) -> [ByteString] -> IO ()
forall a b. (a -> b) -> a -> b
$
          Int -> ByteString -> [ByteString]
forall a. Int -> a -> [a]
replicate Int
burst ByteString
emptySettingsFrame
      TBQueue ByteString -> Socket -> Socket -> Int -> IO ()
readFromClient TBQueue ByteString
toClient Socket
client Socket
upstream Int
sent'

  -- Walk whole HTTP/2 frames out of the buffer, counting the non-ACK SETTINGS
  -- ones, and hand back the trailing partial frame.
  countSettingsFrames :: TVar Int -> ByteString -> IO ByteString
countSettingsFrames TVar Int
settingsSeen ByteString
buffer
    | ByteString -> Int
BS.length ByteString
buffer Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
frameHeaderSize = ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ByteString
buffer
    | Bool
otherwise = do
        let frameSize :: Int
frameSize = Int
frameHeaderSize Int -> Int -> Int
forall a. Num a => a -> a -> a
+ (Int -> Word8 -> Int) -> Int -> ByteString -> Int
forall a. (a -> Word8 -> a) -> a -> ByteString -> a
BS.foldl' (\Int
acc Word8
b -> Int
acc Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
256 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Word8 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word8
b) Int
0 (Int -> ByteString -> ByteString
BS.take Int
3 ByteString
buffer)
            frameType :: Word8
frameType = HasCallStack => ByteString -> Int -> Word8
ByteString -> Int -> Word8
BS.index ByteString
buffer Int
3
            flags :: Word8
flags = HasCallStack => ByteString -> Int -> Word8
ByteString -> Int -> Word8
BS.index ByteString
buffer Int
4
        if ByteString -> Int
BS.length ByteString
buffer Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
frameSize
          then ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ByteString
buffer
          else do
            Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Word8
frameType Word8 -> Word8 -> Bool
forall a. Eq a => a -> a -> Bool
== Word8
settingsFrameType Bool -> Bool -> Bool
&& Bool -> Bool
not (Word8 -> Int -> Bool
forall a. Bits a => a -> Int -> Bool
testBit Word8
flags Int
0)) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
              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 Int -> (Int -> Int) -> STM IO ()
forall a. TVar IO a -> (a -> a) -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar Int
TVar IO Int
settingsSeen (Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)
            TVar Int -> ByteString -> IO ByteString
countSettingsFrames TVar Int
settingsSeen (Int -> ByteString -> ByteString
BS.drop Int
frameSize ByteString
buffer)

  -- SETTINGS, empty payload, no flags, stream 0.
  emptySettingsFrame :: ByteString
emptySettingsFrame = [Word8] -> ByteString
BS.pack [Word8
0, Word8
0, Word8
0, Word8
settingsFrameType, Word8
0, Word8
0, Word8
0, Word8
0, Word8
0]

  settingsFrameType :: Word8
settingsFrameType = Word8
4 :: Word8

  frameHeaderSize :: Int
frameHeaderSize = Int
9

  burstAfterBytes :: Int
burstAfterBytes = Int
16 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1024

  chunkSize :: Int
chunkSize = Int
65536

data PeerConfig2 = PeerConfig2
  { PeerConfig2 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
  , PeerConfig2 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
  }

setup2Peers :: FilePath -> IO PeerConfig2
setup2Peers :: HostName -> IO PeerConfig2
setup2Peers HostName
tmp = do
  -- Allocate peer ports whose derived etcd client ports are also free at
  -- allocation time — otherwise etcd dies on startup with "bind: address
  -- already in use" for the client port. See 'peerPortToClientPort'.
  [PortNumber
port1, PortNumber
port2] <- (Int -> PortNumber) -> [Int] -> [PortNumber]
forall a b. (a -> b) -> [a] -> [b]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Int -> PortNumber
forall a b. (Integral a, Num b) => a -> b
fromIntegral ([Int] -> [PortNumber]) -> IO [Int] -> IO [PortNumber]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (PortNumber -> PortNumber) -> Int -> IO [Int]
randomUnusedTCPPortsWithDerived PortNumber -> PortNumber
peerPortToClientPort Int
2
  let aliceHost :: Host
aliceHost = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
  let bobHost :: Host
bobHost = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
  PeerConfig2 -> IO PeerConfig2
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    PeerConfig2
      { $sel:aliceConfig:PeerConfig2 :: NetworkConfiguration
aliceConfig =
          NetworkConfiguration
            { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
            , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
            , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
            , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = [Party
bob, Party
carol]
            , $sel:peers:NetworkConfiguration :: [Host]
peers = [Host
bobHost]
            , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
            , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
            , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
            }
      , $sel:bobConfig:PeerConfig2 :: NetworkConfiguration
bobConfig =
          NetworkConfiguration
            { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
            , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
            , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
bobSk
            , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = [Party
alice, Party
carol]
            , $sel:peers:NetworkConfiguration :: [Host]
peers = [Host
aliceHost]
            , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"bob"
            , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"bob"
            , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
            }
      }

data PeerConfig3 = PeerConfig3
  { PeerConfig3 -> NetworkConfiguration
aliceConfig :: NetworkConfiguration
  , PeerConfig3 -> NetworkConfiguration
bobConfig :: NetworkConfiguration
  , PeerConfig3 -> NetworkConfiguration
carolConfig :: NetworkConfiguration
  }

setup3Peers :: FilePath -> IO PeerConfig3
setup3Peers :: HostName -> IO PeerConfig3
setup3Peers HostName
tmp = do
  -- See note in 'setup2Peers' about the derived client port.
  [PortNumber
port1, PortNumber
port2, PortNumber
port3] <- (Int -> PortNumber) -> [Int] -> [PortNumber]
forall a b. (a -> b) -> [a] -> [b]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Int -> PortNumber
forall a b. (Integral a, Num b) => a -> b
fromIntegral ([Int] -> [PortNumber]) -> IO [Int] -> IO [PortNumber]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (PortNumber -> PortNumber) -> Int -> IO [Int]
randomUnusedTCPPortsWithDerived PortNumber -> PortNumber
peerPortToClientPort Int
3
  let aliceHost :: Host
aliceHost = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
  let bobHost :: Host
bobHost = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
  let carolHost :: Host
carolHost = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port3
  PeerConfig3 -> IO PeerConfig3
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    PeerConfig3
      { $sel:aliceConfig:PeerConfig3 :: NetworkConfiguration
aliceConfig =
          NetworkConfiguration
            { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
            , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port1
            , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
aliceSk
            , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = [Party
bob, Party
carol]
            , $sel:peers:NetworkConfiguration :: [Host]
peers = [Host
bobHost, Host
carolHost]
            , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"alice"
            , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"alice"
            , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
            }
      , $sel:bobConfig:PeerConfig3 :: NetworkConfiguration
bobConfig =
          NetworkConfiguration
            { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
            , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port2
            , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
bobSk
            , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = [Party
alice, Party
carol]
            , $sel:peers:NetworkConfiguration :: [Host]
peers = [Host
aliceHost, Host
carolHost]
            , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"bob"
            , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"bob"
            , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
            }
      , $sel:carolConfig:PeerConfig3 :: NetworkConfiguration
carolConfig =
          NetworkConfiguration
            { $sel:listen:NetworkConfiguration :: Host
listen = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port3
            , $sel:advertise:NetworkConfiguration :: Host
advertise = Text -> PortNumber -> Host
Host Text
forall s. IsString s => s
lo PortNumber
port3
            , $sel:signingKey:NetworkConfiguration :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
carolSk
            , $sel:otherParties:NetworkConfiguration :: [Party]
otherParties = [Party
alice, Party
bob]
            , $sel:peers:NetworkConfiguration :: [Host]
peers = [Host
aliceHost, Host
bobHost]
            , $sel:nodeId:NetworkConfiguration :: NodeId
nodeId = NodeId
"carol"
            , $sel:persistenceDir:NetworkConfiguration :: HostName
persistenceDir = HostName
tmp HostName -> HostName -> HostName
</> HostName
"carol"
            , $sel:whichEtcd:NetworkConfiguration :: WhichEtcd
whichEtcd = WhichEtcd
SystemEtcd
            }
      }

prop_canRoundtripCBOREncoding ::
  (ToCBOR a, FromCBOR a, Eq a, Show a) => a -> Property
prop_canRoundtripCBOREncoding :: forall a. (ToCBOR a, FromCBOR a, Eq a, Show a) => a -> Property
prop_canRoundtripCBOREncoding a
a =
  let encoded :: ByteString
encoded = Encoding -> ByteString
toLazyByteString (Encoding -> ByteString) -> Encoding -> ByteString
forall a b. (a -> b) -> a -> b
$ a -> Encoding
forall a. ToCBOR a => a -> Encoding
toCBOR a
a
   in ((ByteString, a) -> a
forall a b. (a, b) -> b
snd ((ByteString, a) -> a)
-> Either DeserialiseFailure (ByteString, a)
-> Either DeserialiseFailure a
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (forall s. Decoder s a)
-> ByteString -> Either DeserialiseFailure (ByteString, a)
forall a.
(forall s. Decoder s a)
-> ByteString -> Either DeserialiseFailure (ByteString, a)
deserialiseFromBytes Decoder s a
forall s. Decoder s a
forall a s. FromCBOR a => Decoder s a
fromCBOR ByteString
encoded) Either DeserialiseFailure a
-> Either DeserialiseFailure a -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== a -> Either DeserialiseFailure a
forall a b. b -> Either a b
Right a
a

newRecordingCallback :: MonadLabelledSTM m => m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback :: forall (m :: * -> *) msg.
MonadLabelledSTM m =>
m (NetworkCallback msg m, m msg, m Connectivity)
newRecordingCallback = do
  TQueue m msg
received <- HostName -> m (TQueue m msg)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> m (TQueue m a)
newLabelledTQueueIO HostName
"received"
  TQueue m Connectivity
connectivity <- HostName -> m (TQueue m Connectivity)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
HostName -> m (TQueue m a)
newLabelledTQueueIO HostName
"connectivity"
  (NetworkCallback msg m, m msg, m Connectivity)
-> m (NetworkCallback msg m, m msg, m Connectivity)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    ( NetworkCallback
        { $sel:deliver:NetworkCallback :: msg -> m ()
deliver = 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 ()) -> (msg -> STM m ()) -> msg -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TQueue m msg -> msg -> STM m ()
forall a. TQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> a -> STM m ()
writeTQueue TQueue m msg
received
        , $sel:onConnectivity:NetworkCallback :: Connectivity -> m ()
onConnectivity = 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 ())
-> (Connectivity -> STM m ()) -> Connectivity -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TQueue m Connectivity -> Connectivity -> STM m ()
forall a. TQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> a -> STM m ()
writeTQueue TQueue m Connectivity
connectivity
        }
    , STM m msg -> m msg
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m msg -> m msg) -> STM m msg -> m msg
forall a b. (a -> b) -> a -> b
$ TQueue m msg -> STM m msg
forall a. TQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> STM m a
readTQueue TQueue m msg
received
    , STM m Connectivity -> m Connectivity
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m Connectivity -> m Connectivity)
-> STM m Connectivity -> m Connectivity
forall a b. (a -> b) -> a -> b
$ TQueue m Connectivity -> STM m Connectivity
forall a. TQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> STM m a
readTQueue TQueue m Connectivity
connectivity
    )