natskell 1.2.0.0 → 1.4.0.0

raw patch · 14 files changed

+442/−49 lines, 14 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• client/Client.hs +3/−0
• client/Client/Implementation.hs +24/−4
• internal/Plumbing/Network/Connection/Tls.hs +16/−1
• internal/Policy/Engine.hs +2/−0
• internal/Policy/Handshake/Nats.hs +54/−39
• internal/Policy/State/Types.hs +4/−0
• internal/Types/TLS.hs +49/−2
• jetstream/JetStream/Message.hs +58/−0
• jetstream/JetStream/Message/API.hs +10/−0
• jetstream/JetStream/Message/Types.hs +16/−0
• natskell.cabal +1/−1
• test/PublicAPI/Main.hs +28/−0
• test/Unit/ConnectionSpec.hs +61/−2
• test/Unit/JetStream/MessageSpec.hs +116/−0

@@ -24,6 +24,7 @@   , withTLSRootCA   , withTLSServerName   , withTLSInsecure+  , withTLSConfigSource   , withMinimumLogLevel   , withLogAction   , withConnectionAttempts@@ -53,6 +54,8 @@   , TLSPrivateKey   , TLSCertData   , TLSConfig (..)+  , TLSConfigSource+  , TLSConfigSourceFailure (..)   , ClientExitReason (..)   , ConnectionEvent (..)   , ServerError

@@ -27,6 +27,7 @@   , withTLSRootCA   , withTLSServerName   , withTLSInsecure+  , withTLSConfigSource   , withMinimumLogLevel   , withLogAction   , withConnectionAttempts@@ -56,6 +57,8 @@   , TLSPrivateKey   , TLSCertData   , TLSConfig (..)+  , TLSConfigSource+  , TLSConfigSourceFailure (..)   , ClientExitReason (..)   , ConnectionEvent (..)   , ServerError@@ -194,6 +197,8 @@     , ServerErrorKind (..)     , TLSCertData     , TLSConfig (..)+    , TLSConfigSource+    , TLSConfigSourceFailure (..)     , TLSPrivateKey     , TLSPublicKey     , defaultTLSConfig@@ -235,6 +240,7 @@                        { optionConnectConfig :: Connect.Connect                        , optionAuth :: Auth                        , optionTlsConfig :: Maybe TLSConfig+                       , optionTlsConfigSource :: Maybe TLSConfigSource                        , optionLoggerConfig :: LoggerConfig                        , optionConnectionAttempts :: Int                        , optionConnectTimeoutMicros :: Int@@ -294,6 +300,7 @@         { optionConnectConfig = defaultConnect         , optionAuth = AuthNone.auth         , optionTlsConfig = Nothing+        , optionTlsConfigSource = Nothing         , optionLoggerConfig = loggerConfig'         , optionConnectionAttempts = 5         , optionConnectTimeoutMicros = 2 * 1000000@@ -315,6 +322,7 @@           , connectConfig = optionConnectConfig defaultOptions           , loggerConfig = optionLoggerConfig defaultOptions           , tlsConfig = optionTlsConfig defaultOptions+          , tlsConfigSource = optionTlsConfigSource defaultOptions           , serverErrorHandler = optionServerErrorHandler defaultOptions           , connectionEventHandler = optionConnectionEventHandler defaultOptions           , exitAction = optionExitAction defaultOptions@@ -533,11 +541,23 @@ withTLSInsecure =   modifyTLSConfig $ \tls -> tls { tlsInsecure = True } +-- | Fetch complete mutual-TLS configuration before every TLS handshake.+--+-- This is intended for renewable identities such as SPIFFE SVIDs. It replaces+-- static TLS options; applying a static TLS option afterwards replaces source.+withTLSConfigSource :: TLSConfigSource -> ConfigOption+withTLSConfigSource source config =+  config+    { optionTlsConfig = Nothing+    , optionTlsConfigSource = Just source+    }+ modifyTLSConfig :: (TLSConfig -> TLSConfig) -> ConfigOption modifyTLSConfig update config =   config     { optionTlsConfig =         Just (update (fromMaybe defaultTLSConfig (optionTlsConfig config)))+    , optionTlsConfigSource = Nothing     }  withMinimumLogLevel :: LogLevel -> ConfigOption@@ -659,10 +679,10 @@       [] -> logMessage Info "no authentication method provided"       methods -> forM_ methods $ \method ->         logMessage Info ("using " ++ method)-    case optionTlsConfig options of-      Nothing ->-        pure ()-      Just tls -> do+    case (optionTlsConfig options, optionTlsConfigSource options) of+      (Nothing, Nothing) -> pure ()+      (Nothing, Just _) -> logMessage Info "using renewable tls configuration"+      (Just tls, _) -> do         logMessage Info "using tls"         when (tlsInsecure tls) $           logMessage Warn "tls certificate verification is disabled"

@@ -1,10 +1,12 @@ module Network.Connection.Tls   ( configureTransport+  , receiveExactly   , upgradeTcp   ) where  import           Control.Exception import           Control.Monad+import qualified Data.ByteString            as BS import qualified Data.ByteString.Char8      as BC import qualified Data.ByteString.Lazy       as LBS import           Data.Maybe                 (fromMaybe)@@ -46,7 +48,7 @@ upgradeTcp sock params = mask $ \restore -> do   let backend = TLS.Backend         { TLS.backendSend = NSB.sendAll sock-        , TLS.backendRecv = NSB.recv sock+        , TLS.backendRecv = receiveExactly (NSB.recv sock)         , TLS.backendFlush = pure ()         , TLS.backendClose = NS.close sock         }@@ -58,6 +60,19 @@   case result of     Left err        -> return $ Left (show err)     Right transport -> return $ Right transport++receiveExactly :: (Int -> IO BS.ByteString) -> Int -> IO BS.ByteString+receiveExactly receive =+  go []+  where+    go chunks remaining+      | remaining <= 0 =+          pure (BS.concat (reverse chunks))+      | otherwise = do+          chunk <- receive remaining+          if BS.null chunk+            then pure (BS.concat (reverse chunks))+            else go (chunk : chunks) (remaining - BS.length chunk)  upgradeToTLS :: Conn -> TLS.ClientParams -> IO (Either String ()) upgradeToTLS conn params = mask_ $ do

@@ -321,11 +321,13 @@      handshakeFailure (HandshakeTransportError err) = ConnectTransportFailure err     handshakeFailure (HandshakeTLSError err) = ConnectTLSFailure err+    handshakeFailure (HandshakeTLSConfigSourceError err) = ConnectTLSConfigSourceFailure err     handshakeFailure (HandshakeProtocolError err) = ConnectProtocolFailure err     handshakeFailure (HandshakeAuthError err) = ConnectAuthenticationFailure (show err)      failureCategory (ConnectTransportFailure _)      = "transport failure"     failureCategory (ConnectTLSFailure _)            = "tls failure"+    failureCategory (ConnectTLSConfigSourceFailure _) = "tls configuration source failure"     failureCategory (ConnectProtocolFailure _)       = "protocol failure"     failureCategory (ConnectAuthenticationFailure _) = "authentication failure"     failureCategory ConnectHandshakeTimeout          = "handshake timeout"

@@ -42,10 +42,15 @@ import qualified Types.Info                as I import           Types.Ping                (Ping (Ping)) import           Types.Pong                (Pong (Pong))-import           Types.TLS                 (defaultTLSConfig)+import           Types.TLS+    ( TLSConfigSourceFailure+    , defaultTLSConfig+    , readTLSConfigSource+    )  data HandshakeError = HandshakeTransportError String                     | HandshakeTLSError String+                    | HandshakeTLSConfigSourceError TLSConfigSourceFailure                     | HandshakeProtocolError String                     | HandshakeAuthError AuthError   deriving (Eq, Show)@@ -63,53 +68,63 @@           pure (Left err)         Right (info, rest) -> do           let cfg = config state-              tlsRequested = isJust (tlsConfig cfg) || Connect.tls_required (connectConfig cfg)+              tlsRequested =+                isJust (tlsConfig cfg)+                  || isJust (tlsConfigSource cfg)+                  || Connect.tls_required (connectConfig cfg)               tlsRequired = fromMaybe False (I.tls_required info)               tlsAvailable = fromMaybe False (I.tls_available info)               useTls = tlsRequired || (tlsRequested && tlsAvailable)-              transportOption =-                TransportOption-                  { transportHost = host-                  , transportTlsRequired = tlsRequired-                  , transportTlsRequested = tlsRequested && tlsAvailable-                  , transportTlsConfig =-                      if useTls-                        then Just (fromMaybe defaultTLSConfig (tlsConfig cfg))-                        else Nothing-                  , transportInitialBytes = rest-                  }           if tlsRequested && not (tlsRequired || tlsAvailable)             then pure (Left (HandshakeTLSError "server does not offer TLS"))             else do-              transportResult <- configure connectionApi conn transportOption-              case transportResult of-                Left err ->-                  pure (Left (HandshakeTLSError err))-                Right () -> do-                  setServerInfo state info-                  updateLogContextFromInfo state info-                  let authContext = AuthContext { authNonce = I.nonce info }-                      connectPayload =-                        (connectConfig cfg)-                          { Connect.tls_required = useTls+              tlsResult <- resolveTlsConfig cfg useTls+              case tlsResult of+                Left err -> pure (Left (HandshakeTLSConfigSourceError err))+                Right tls -> do+                  let transportOption =+                        TransportOption+                          { transportHost = host+                          , transportTlsRequired = tlsRequired+                          , transportTlsRequested = tlsRequested && tlsAvailable+                          , transportTlsConfig = tls+                          , transportInitialBytes = rest                           }-                  authResult <- buildAuthPatch auth authContext-                  case authResult of+                  transportResult <- configure connectionApi conn transportOption+                  case transportResult of                     Left err ->-                      pure (Left (HandshakeAuthError err))-                    Right patch -> do-                      writeResult <--                        writeDataLazy-                          (writer connectionApi)-                          conn-                          ( transform (applyAuthPatch patch connectPayload)-                              <> transform Ping-                          )-                      case writeResult of+                      pure (Left (HandshakeTLSError err))+                    Right () -> do+                      setServerInfo state info+                      updateLogContextFromInfo state info+                      let authContext = AuthContext { authNonce = I.nonce info }+                          connectPayload =+                            (connectConfig cfg)+                              { Connect.tls_required = useTls+                              }+                      authResult <- buildAuthPatch auth authContext+                      case authResult of                         Left err ->-                          pure (Left (HandshakeTransportError err))-                        Right () ->-                          awaitPong mempty+                          pure (Left (HandshakeAuthError err))+                        Right patch -> do+                          writeResult <-+                            writeDataLazy+                              (writer connectionApi)+                              conn+                              ( transform (applyAuthPatch patch connectPayload)+                                  <> transform Ping+                              )+                          case writeResult of+                            Left err ->+                              pure (Left (HandshakeTransportError err))+                            Right () ->+                              awaitPong mempty++    resolveTlsConfig _ False = pure (Right Nothing)+    resolveTlsConfig cfg True =+      case tlsConfigSource cfg of+        Just source -> fmap Just <$> readTLSConfigSource source+        Nothing -> pure (Right (Just (fromMaybe defaultTLSConfig (tlsConfig cfg))))      readInitialInfo :: IO (Either HandshakeError (I.Info, BS.ByteString))     readInitialInfo = readMore mempty

@@ -5,6 +5,8 @@   , TLSPrivateKey   , TLSCertData   , TLSConfig (..)+  , TLSConfigSource+  , TLSConfigSourceFailure (..)   , defaultTLSConfig   , ClientConfig (..)   , ConnectError (..)@@ -35,6 +37,7 @@                       , connectConfig          :: Connect                       , loggerConfig           :: LoggerConfig                       , tlsConfig              :: Maybe TLSConfig+                      , tlsConfigSource        :: Maybe TLSConfigSource                       , serverErrorHandler     :: ServerError -> IO ()                       , connectionEventHandler :: ConnectionEvent -> IO ()                       , exitAction             :: ClientExitReason -> IO ()@@ -56,6 +59,7 @@ -- | The stage at which a connection attempt failed. data ConnectFailure = ConnectTransportFailure String                     | ConnectTLSFailure String+                    | ConnectTLSConfigSourceFailure TLSConfigSourceFailure                     | ConnectProtocolFailure String                     | ConnectAuthenticationFailure String                     | ConnectHandshakeTimeout

@@ -3,11 +3,21 @@   , TLSPrivateKey   , TLSCertData   , TLSConfig (..)+  , TLSConfigSource+  , TLSConfigSourceFailure (..)   , defaultTLSConfig+  , readTLSConfigSource   ) where -import qualified Data.ByteString as BS-import           Data.Maybe      (isJust)+import           Control.Exception+    ( SomeAsyncException+    , SomeException+    , fromException+    , throwIO+    , try+    )+import qualified Data.ByteString   as BS+import           Data.Maybe        (isJust)  type TLSPublicKey = BS.ByteString type TLSPrivateKey = BS.ByteString@@ -22,6 +32,19 @@                    }   deriving (Eq) +-- | Reads a complete mutual-TLS configuration snapshot for one connection.+--+-- Sources are invoked immediately before each TLS handshake, including+-- reconnects and client resets.+type TLSConfigSource = IO (Either TLSConfigSourceFailure TLSConfig)++-- | Stable failure categories for renewable TLS configuration.+--+-- Raw source exceptions are intentionally not exposed because they can contain+-- certificate paths, secret-manager responses, or other private details.+data TLSConfigSourceFailure = TLSConfigSourceUnavailable | TLSConfigSourceIncomplete+  deriving (Eq, Show)+ instance Show TLSConfig where   show config =     "TLSConfig {tlsClientCertificate = "@@ -45,3 +68,27 @@     , tlsServerName = Nothing     , tlsInsecure = False     }++-- | Read and validate a source snapshot without leaking source exceptions.+--+-- Mutual-TLS sources must include their complete trust and identity material.+-- This prevents a certificate/key rotation from accidentally combining files+-- from different SVID generations with static client options.+readTLSConfigSource :: TLSConfigSource -> IO (Either TLSConfigSourceFailure TLSConfig)+readTLSConfigSource source = do+  result <- try source :: IO (Either SomeException (Either TLSConfigSourceFailure TLSConfig))+  case result of+    Left err ->+      case fromException err :: Maybe SomeAsyncException of+        Just _  -> throwIO err+        Nothing -> pure (Left TLSConfigSourceUnavailable)+    Right (Left err) -> pure (Left err)+    Right (Right config)+      | complete config -> pure (Right config)+      | otherwise       -> pure (Left TLSConfigSourceIncomplete)+  where+    complete config =+      isJust (tlsClientCertificate config)+        && not (null (tlsRootCertificates config))+        && isJust (tlsServerName config)+        && not (tlsInsecure config)

@@ -10,6 +10,7 @@   ) where  import qualified Client.API                 as Nats+import           Control.Concurrent.MVar    (modifyMVar, newMVar) import           Control.Concurrent.STM import           Control.Exception          (bracket) import           Control.Monad              (unless, void)@@ -69,6 +70,7 @@   MessageAPI     { fetch = \stream consumer options requestOptions ->         fetchMessages context stream consumer (pullRequest options) requestOptions+    , consumePull = consumePullMessages context     , consumePush = \deliverSubject options _ handler ->         consumePushMessages context deliverSubject (pushConsumeConfig options) handler     , createOrderedConsumer = \stream options requestOptions ->@@ -82,6 +84,62 @@     , term = sendDisposition context ConfirmWhenRequested (ackPayload Term)     , termSync = sendDisposition context ConfirmAlways (ackPayload Term)     }++consumePullMessages+  :: JetStreamContext+  -> StreamName+  -> ConsumerName+  -> (PullEvent -> IO ())+  -> IO (Either JetStreamError PullSubscription)+consumePullMessages context stream consumer handler = do+  inbox <- Nats.newInbox natsClient+  subscription <- Nats.subscribe natsClient inbox [] $ \msgView ->+    let message = fromMsgView msgView+    in case messageStatus message of+         Nothing     -> handler (PullMessage message)+         Just status -> handler (PullStatusMessage status)+  case subscription of+    Left err ->+      pure (Left (JetStreamNatsError err))+    Right handle -> do+      stopped <- newMVar False+      pure $ Right PullSubscription+        { requestPull = \options ->+            let request = pullRequest options+            in modifyMVar stopped $ \isStopped ->+                 if isStopped+                   then pure (True, Left JetStreamNoReply)+                   else do+                     result <- requestPullMessages context stream consumer inbox request+                     pure (False, result)+        , stopPullSubscription =+            modifyMVar stopped $ \isStopped ->+              if isStopped+                then pure (True, Right ())+                else do+                  result <- mapNatsResult <$> Nats.unsubscribe natsClient handle []+                  pure (True, result)+        }+  where+    natsClient = contextClient context++requestPullMessages+  :: JetStreamContext+  -> StreamName+  -> ConsumerName+  -> Subject+  -> PullRequest+  -> IO (Either JetStreamError ())+requestPullMessages context stream consumer inbox request+  | pullRequestBatch request <= 0 =+      pure (Right ())+  | otherwise =+      mapNatsResult <$>+        Nats.publish+          (contextClient context)+          (Subject.consumerNextSubject context stream consumer)+          (pullRequestPayload (pullRequestBatch request) request)+          [Nats.withReplyTo inbox]  consumePushMessages   :: JetStreamContext

@@ -1,6 +1,7 @@ module JetStream.Message.API   ( MessageAPI   , fetch+  , consumePull   , consumePush   , createOrderedConsumer   , ack@@ -39,6 +40,10 @@   , PushConsumeOption   , PushSubscription   , stopPushSubscription+  , PullEvent (..)+  , PullSubscription+  , requestPull+  , stopPullSubscription   , PullResponse   , pullResponseMessages   , pullResponseStatus@@ -64,12 +69,15 @@     , NakDelay     , OrderedConsumer     , OrderedConsumerOption+    , PullEvent (..)     , PullResponse     , PullStatus (..)+    , PullSubscription     , PushConsumeOption     , PushSubscription     , ack     , ackSync+    , consumePull     , consumePush     , createOrderedConsumer     , fetch@@ -95,7 +103,9 @@     , orderedConsumerInfo     , pullResponseMessages     , pullResponseStatus+    , requestPull     , stopOrderedConsumer+    , stopPullSubscription     , stopPushSubscription     , term     , termSync

@@ -15,8 +15,10 @@   , PushConsumeConfig (..)   , PushConsumeOption   , PushSubscription (..)+  , PullEvent (..)   , PullRequest (..)   , PullResponse (..)+  , PullSubscription (..)   , PullStatus (..)   , ackPayload   , classifyStatusHeaders@@ -73,6 +75,7 @@ -- hides the constructor so this capability can grow additively. data MessageAPI = MessageAPI                     { fetch :: StreamName -> ConsumerName -> [FetchOption] -> [JetStreamRequestOption] -> IO (Either JetStreamError PullResponse)+                    , consumePull :: StreamName -> ConsumerName -> (PullEvent -> IO ()) -> IO (Either JetStreamError PullSubscription)                     , consumePush :: Subject -> [PushConsumeOption] -> [JetStreamRequestOption] -> (Message -> IO ()) -> IO (Either JetStreamError PushSubscription)                     , createOrderedConsumer :: StreamName -> [OrderedConsumerOption] -> [JetStreamRequestOption] -> IO (Either JetStreamError OrderedConsumer)                     , ack :: Message -> [JetStreamRequestOption] -> IO (Either JetStreamError ())@@ -121,6 +124,12 @@                       }   deriving (Eq, Show) +-- | One reply received by a persistent pull subscription. Status replies end+-- or reject a server-side pull request without being application messages.+data PullEvent = PullMessage Message+               | PullStatusMessage PullStatus+  deriving (Eq, Show)+ data Message = Message                  { messageSubject :: Subject                  , messagePayload :: Payload@@ -143,6 +152,13 @@   deriving (Eq, Show)  newtype PushSubscription = PushSubscription { stopPushSubscription :: IO (Either JetStreamError ()) }++-- | One reusable reply inbox for explicit pull requests. Requests may select+-- different batch sizes while the subscription remains active.+data PullSubscription = PullSubscription+                          { requestPull :: [FetchOption] -> IO (Either JetStreamError ())+                          , stopPullSubscription :: IO (Either JetStreamError ())+                          }  newtype PushConsumeConfig = PushConsumeConfig { pushConsumeQueueGroup :: Maybe Subject } 

@@ -1,6 +1,6 @@ cabal-version:          3.0 name:                   natskell-version:                1.2.0.0+version:                1.4.0.0 synopsis:               A NATS client library written in Haskell tested-with:   GHC ==8.8,

@@ -35,6 +35,12 @@     Left err       -> error (show err)     Right endpoint -> endpoint +tlsSource :: Client.TLSConfigSource+tlsSource = pure (Left Client.TLSConfigSourceUnavailable)++tlsSourceOption :: Client.ConfigOption+tlsSourceOption = Client.withTLSConfigSource tlsSource+ serverBuilders   :: ( Either Client.ServerConfigError Client.Server      , Either Client.ServerConfigError Client.Server@@ -168,6 +174,19 @@     "processor"     [JetStreamMessage.withFetchBatch 10]     []+  pullSubscription <- JetStreamMessage.consumePull+    (JetStream.messages jetStream)+    "ORDERS"+    "processor"+    observePullEvent+  case pullSubscription of+    Left _ -> pure ()+    Right handle -> do+      _ <- JetStreamMessage.requestPull+        handle+        [JetStreamMessage.withFetchBatch 10]+      _ <- JetStreamMessage.stopPullSubscription handle+      pure ()   _ <- JetStreamMessage.consumePush     (JetStream.messages jetStream)     "deliver.orders"@@ -212,6 +231,14 @@           pure ()   pure () +observePullEvent :: JetStreamMessage.PullEvent -> IO ()+observePullEvent event =+  case event of+    JetStreamMessage.PullMessage message ->+      JetStreamMessage.messagePayload message `seq` pure ()+    JetStreamMessage.PullStatusMessage status ->+      status `seq` pure ()+ messageOperations   :: JetStream.MessageAPI   -> JetStreamMessage.Message@@ -245,4 +272,5 @@     `seq` jetStreamOperations     `seq` messageOperations     `seq` inspectJetStreamApiError+    `seq` tlsSourceOption     `seq` pure ()

@@ -11,9 +11,10 @@ import           Control.Monad             (replicateM_, void, when) import qualified Data.ByteString           as BS import qualified Data.ByteString.Lazy      as LBS+import           Data.IORef                (newIORef, readIORef, writeIORef) import           Engine                    (runEngine) import           Handshake.Nats-    ( HandshakeError (HandshakeAuthError, HandshakeProtocolError, HandshakeTLSError)+    ( HandshakeError (HandshakeAuthError, HandshakeProtocolError, HandshakeTLSConfigSourceError, HandshakeTLSError)     , performHandshake     ) import           Lib.Logger@@ -24,6 +25,7 @@     ) import           Network.Connection        (connectionApi) import           Network.Connection.Core   (Transport (..), pointTransport)+import           Network.Connection.Tls    (receiveExactly) import           Network.ConnectionAPI     ( ConnectionAPI (..)     , ReaderAPI (..)@@ -102,7 +104,12 @@ import           Types.Ping                (Ping (Ping)) import           Types.Pong                (Pong (Pong)) import qualified Types.Pub                 as Pub-import           Types.TLS                 (TLSConfig (..), defaultTLSConfig)+import           Types.TLS+    ( TLSConfig (..)+    , TLSConfigSourceFailure (..)+    , defaultTLSConfig+    , readTLSConfigSource+    )  spec :: Spec spec = do@@ -175,6 +182,35 @@       show config `shouldNotContain` "private-root"       show config `shouldContain` "tlsRootCertificates = 1 configured" +    it "rejects incomplete renewable TLS snapshots" $ do+      readTLSConfigSource (pure (Right defaultTLSConfig))+        `shouldReturn` Left TLSConfigSourceIncomplete++    it "redacts renewable TLS source exceptions" $ do+      readTLSConfigSource (throwIO (userError "private certificate path"))+        `shouldReturn` Left TLSConfigSourceUnavailable++    it "fills each TLS backend receive across partial socket reads" $ do+      chunks <- newMVar ["ab", "c", "de"]+      requested <- newMVar []+      let receive count = do+            modifyMVar_ requested (pure . (<> [count]))+            modifyMVar chunks $ \case+              []           -> pure ([], BS.empty)+              next : rest -> pure (rest, next)++      receiveExactly receive 5 `shouldReturn` "abcde"+      readMVar requested `shouldReturn` [5, 3, 2]++    it "returns a partial receive only when the socket reaches EOF" $ do+      chunks <- newMVar ["ab"]+      let receive _ =+            modifyMVar chunks $ \case+              []           -> pure ([], BS.empty)+              next : rest -> pure (rest, next)++      receiveExactly receive 5 `shouldReturn` "ab"+   describe "Connection reader" $ do     it "unblocks a blocking read when closeReader is called" $ do       conn <- newConn connectionApi@@ -971,6 +1007,24 @@         Left (HandshakeTLSError _) -> pure ()         other -> expectationFailure ("expected TLS error, got: " ++ show other) +    it "reads renewable TLS configuration before each TLS handshake" $ do+      sourceCalls <- newIORef (0 :: Int)+      state <- newTestStateWithConfig $ \cfg ->+        cfg+          { tlsConfigSource = Just $ do+              writeIORef sourceCalls 1+              pure (Left TLSConfigSourceUnavailable)+          }+      conn <- newConn connectionApi+      writes <- newTVarIO []+      transport <- newScriptedTransport [tlsInfoFrame] writes+      pointTransport conn transport++      result <- performHandshake connectionApi Attoparsec.parserApi state auth conn "127.0.0.1"++      result `shouldBe` Left (HandshakeTLSConfigSourceError TLSConfigSourceUnavailable)+      readIORef sourceCalls `shouldReturn` 1+ newTestState = do   newTestStateWithConfig id @@ -1011,6 +1065,7 @@     , connectConfig = testConnect     , loggerConfig = logger     , tlsConfig = Nothing+    , tlsConfigSource = Nothing     , serverErrorHandler = const (pure ())     , connectionEventHandler = const (pure ())     , exitAction = const (pure ())@@ -1114,6 +1169,10 @@ infoFrame :: BS.ByteString infoFrame =   "INFO {\"server_id\":\"srv\",\"version\":\"1.0.0\",\"go\":\"go1\",\"host\":\"127.0.0.1\",\"port\":4222,\"max_payload\":1024,\"proto\":1}\r\n"++tlsInfoFrame :: BS.ByteString+tlsInfoFrame =+  "INFO {\"server_id\":\"srv\",\"version\":\"1.0.0\",\"go\":\"go1\",\"host\":\"127.0.0.1\",\"port\":4222,\"max_payload\":1024,\"proto\":1,\"tls_available\":true}\r\n"  newScriptedTransport :: [BS.ByteString] -> TVar [BS.ByteString] -> IO Transport newScriptedTransport chunks writes = do

@@ -228,6 +228,63 @@       unsubscribeCalls' <- readIORef unsubscribeCalls       unsubscribeCalls' `shouldBe` ["sid-timeout"] +  describe "persistent pull" $ do+    it "reuses one reply subscription across batched pull requests" $ do+      publishCalls <- newIORef []+      subscribeCalls <- newIORef []+      unsubscribeCalls <- newIORef []+      callbackRef <- newIORef Nothing+      events <- newIORef []+      let client = persistentPullFakeClient+            publishCalls+            subscribeCalls+            unsubscribeCalls+            callbackRef+          api = messageAPI+            (newJetStreamContext client [])+            (error "unused consumer API")++      subscriptionResult <- consumePull api "ORDERS" "WORKER" $ \event ->+        modifyIORef' events (++ [event])+      subscription <-+        case subscriptionResult of+          Left err ->+            expectationFailure ("pull subscription failed: " ++ show err) >>+              fail "pull subscription failed"+          Right value ->+            pure value++      requestPull subscription [withFetchBatch 2] `shouldReturn` Right ()+      requestPull+        subscription+        [ withFetchBatch 3+        , withFetchWait (FetchNoWaitMicros 1000)+        ]+        `shouldReturn` Right ()++      readIORef subscribeCalls `shouldReturn` ["_INBOX.persistent"]+      readIORef publishCalls `shouldReturn`+        [ ( "$JS.API.CONSUMER.MSG.NEXT.ORDERS.WORKER"+          , "{\"batch\":2,\"expires\":1000000000}"+          , Just "_INBOX.persistent"+          )+        , ( "$JS.API.CONSUMER.MSG.NEXT.ORDERS.WORKER"+          , "{\"batch\":3,\"no_wait\":true}"+          , Just "_INBOX.persistent"+          )+        ]+      readIORef events `shouldReturn`+        [ PullMessage (fromFakeMsg (fakeMsg "one"))+        , PullMessage (fromFakeMsg (fakeMsg "two"))+        , PullStatusMessage (PullNoMessages (Just "No Messages"))+        ]++      stopPullSubscription subscription `shouldReturn` Right ()+      stopPullSubscription subscription `shouldReturn` Right ()+      readIORef unsubscribeCalls `shouldReturn` ["sid-persistent"]+      requestPull subscription [withFetchBatch 1]+        `shouldReturn` Left JetStreamNoReply+ fakeClient   :: IORef [(Subject, Payload, Maybe Subject)]   -> IORef [SID]@@ -270,6 +327,65 @@     , Nats.replyTo = Just "$JS.ACK.ORDERS.WORKER.1.1.1"     , Nats.payload = body     , Nats.headers = Nothing+    }++fromFakeMsg :: Nats.Message -> Message+fromFakeMsg value =+  Message+    { messageSubject = Nats.subject value+    , messagePayload = Nats.payload value+    , messageHeaders = Nats.headers value+    , messageReplyTo = Nats.replyTo value+    , messageStatus = classifyStatusHeaders (Nats.headers value)+    }++persistentPullFakeClient+  :: IORef [(Subject, Payload, Maybe Subject)]+  -> IORef [Subject]+  -> IORef [SID]+  -> IORef (Maybe (Nats.MsgView -> IO ()))+  -> Nats.Client+persistentPullFakeClient publishCalls subscribeCalls unsubscribeCalls callbackRef =+  Nats.Client+    { Nats.publish = \subject body options -> do+        let replyTo' = publishReplyTo (foldl (flip ($)) defaultPublishConfig options)+        modifyIORef' publishCalls (++ [(subject, body, replyTo')])+        callback <- readIORef callbackRef+        case callback of+          Nothing ->+            pure ()+          Just deliver+            | body == "{\"batch\":2,\"expires\":1000000000}" -> do+                deliver (fakeMsg "one")+                deliver (fakeMsg "two")+            | otherwise ->+                deliver+                  Nats.Message+                    { Nats.subject = "_INBOX.persistent"+                    , Nats.sid = "sid-persistent"+                    , Nats.replyTo = Nothing+                    , Nats.payload = ""+                    , Nats.headers = Just+                        [ ("Status", "404")+                        , ("Description", "No Messages")+                        ]+                    }+        pure (Right ())+    , Nats.subscribe = \subject _ callback -> do+        modifyIORef' subscribeCalls (++ [subject])+        writeIORef callbackRef (Just callback)+        pure (Right (Nats.Subscription "sid-persistent"))+    , Nats.subscribeOnce = \_ _ _ -> pure (Right (Nats.Subscription "sid-once"))+    , Nats.request = \_ _ _ -> pure (Left Nats.NatsRequestTimedOut)+    , Nats.unsubscribe = \(Nats.Subscription sid) _ -> do+        modifyIORef' unsubscribeCalls (++ [sid])+        pure (Right ())+    , Nats.newInbox = pure "_INBOX.persistent"+    , Nats.ping = \_ -> pure (Right ())+    , Nats.flush = \_ -> pure (Right ())+    , Nats.connectionState = pure Nats.ConnectionConnected+    , Nats.reset = \_ -> pure ()+    , Nats.close = \_ -> pure ()     }  silentFakeClient