diff --git a/client/Client.hs b/client/Client.hs --- a/client/Client.hs +++ b/client/Client.hs @@ -24,6 +24,7 @@ , withTLSRootCA , withTLSServerName , withTLSInsecure + , withTLSConfigSource , withMinimumLogLevel , withLogAction , withConnectionAttempts @@ -53,6 +54,8 @@ , TLSPrivateKey , TLSCertData , TLSConfig (..) + , TLSConfigSource + , TLSConfigSourceFailure (..) , ClientExitReason (..) , ConnectionEvent (..) , ServerError diff --git a/client/Client/Implementation.hs b/client/Client/Implementation.hs --- a/client/Client/Implementation.hs +++ b/client/Client/Implementation.hs @@ -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" diff --git a/internal/Plumbing/Network/Connection/Tls.hs b/internal/Plumbing/Network/Connection/Tls.hs --- a/internal/Plumbing/Network/Connection/Tls.hs +++ b/internal/Plumbing/Network/Connection/Tls.hs @@ -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 diff --git a/internal/Policy/Engine.hs b/internal/Policy/Engine.hs --- a/internal/Policy/Engine.hs +++ b/internal/Policy/Engine.hs @@ -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" diff --git a/internal/Policy/Handshake/Nats.hs b/internal/Policy/Handshake/Nats.hs --- a/internal/Policy/Handshake/Nats.hs +++ b/internal/Policy/Handshake/Nats.hs @@ -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 diff --git a/internal/Policy/State/Types.hs b/internal/Policy/State/Types.hs --- a/internal/Policy/State/Types.hs +++ b/internal/Policy/State/Types.hs @@ -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 diff --git a/internal/Types/TLS.hs b/internal/Types/TLS.hs --- a/internal/Types/TLS.hs +++ b/internal/Types/TLS.hs @@ -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) diff --git a/jetstream/JetStream/Message.hs b/jetstream/JetStream/Message.hs --- a/jetstream/JetStream/Message.hs +++ b/jetstream/JetStream/Message.hs @@ -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 diff --git a/jetstream/JetStream/Message/API.hs b/jetstream/JetStream/Message/API.hs --- a/jetstream/JetStream/Message/API.hs +++ b/jetstream/JetStream/Message/API.hs @@ -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 diff --git a/jetstream/JetStream/Message/Types.hs b/jetstream/JetStream/Message/Types.hs --- a/jetstream/JetStream/Message/Types.hs +++ b/jetstream/JetStream/Message/Types.hs @@ -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 } diff --git a/natskell.cabal b/natskell.cabal --- a/natskell.cabal +++ b/natskell.cabal @@ -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, diff --git a/test/PublicAPI/Main.hs b/test/PublicAPI/Main.hs --- a/test/PublicAPI/Main.hs +++ b/test/PublicAPI/Main.hs @@ -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 () diff --git a/test/Unit/ConnectionSpec.hs b/test/Unit/ConnectionSpec.hs --- a/test/Unit/ConnectionSpec.hs +++ b/test/Unit/ConnectionSpec.hs @@ -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 diff --git a/test/Unit/JetStream/MessageSpec.hs b/test/Unit/JetStream/MessageSpec.hs --- a/test/Unit/JetStream/MessageSpec.hs +++ b/test/Unit/JetStream/MessageSpec.hs @@ -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