freckle-app 1.9.3.0 → 1.9.4.0
raw patch · 7 files changed
+96/−39 lines, 7 files
Files
- CHANGELOG.md +6/−1
- freckle-app.cabal +2/−1
- library/Freckle/App/Async.hs +56/−0
- library/Freckle/App/Bugsnag.hs +8/−5
- library/Freckle/App/Kafka/Consumer.hs +15/−24
- library/Freckle/App/Kafka/Producer.hs +8/−7
- package.yaml +1/−1
CHANGELOG.md view
@@ -1,4 +1,9 @@-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.9.1.1...main)+## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.9.4.0...main)++## [v1.9.4.0](https://github.com/freckle/freckle-app/compare/v1.9.3.0...v1.9.4.0)++- Add `Freckle.App.Async` (primarily) for correctly managing thread context when+ spawning new threads. ## [v1.9.3.0](https://github.com/freckle/freckle-app/compare/v1.9.2.1...v1.9.3.0)
freckle-app.cabal view
@@ -1,6 +1,6 @@ cabal-version: 1.18 name: freckle-app-version: 1.9.3.0+version: 1.9.4.0 license: MIT license-file: LICENSE maintainer: Freckle Education@@ -24,6 +24,7 @@ Configuration.Dotenv.Compat Freckle.App Freckle.App.Aeson+ Freckle.App.Async Freckle.App.Bugsnag Freckle.App.Bugsnag.MetaData Freckle.App.Csv
+ library/Freckle/App/Async.hs view
@@ -0,0 +1,56 @@+module Freckle.App.Async+ ( async+ , immortalCreate+ , immortalCreateLogged+ )+where++import Freckle.App.Prelude++import Blammo.Logging+import qualified Control.Immortal as Immortal+import Control.Monad (forever)+import qualified Data.Aeson.Compat as KeyMap+import UnliftIO.Async (Async)+import qualified UnliftIO.Async as UnliftIO+import UnliftIO.Concurrent (threadDelay)+import UnliftIO.Exception (SomeException, displayException)++-- | 'UnliftIO.Async.async' but passing the thread context along+async :: (MonadMask m, MonadUnliftIO m) => m a -> m (Async a)+async f = do+ tc <- liftIO $ KeyMap.toList <$> myThreadContext+ UnliftIO.async $ withThreadContext tc f++-- | Wrapper around creating "Control.Immortal" processes+--+-- Features:+--+-- - Ensures the thread context is correctly passed to both your spawned action+-- and your error handler+-- - Blocks forever after spawning your thread.+immortalCreate+ :: (MonadMask m, MonadUnliftIO m)+ => (Either SomeException () -> m ())+ -- ^ How to handle unexpected finish+ -> m ()+ -- ^ The action to run persistently+ -> m a+immortalCreate onUnexpected act = do+ tc <- liftIO $ KeyMap.toList <$> myThreadContext++ let+ act' = withThreadContext tc act+ onUnexpected' = withThreadContext tc . onUnexpected++ void $ Immortal.create $ \thread -> do+ Immortal.onUnexpectedFinish thread onUnexpected' act'++ forever $ threadDelay maxBound++-- | 'immortalCreate' with logging of unexpected finishes+immortalCreateLogged+ :: (MonadMask m, MonadUnliftIO m, MonadLogger m) => m () -> m a+immortalCreateLogged = immortalCreate $ either logEx pure+ where+ logEx ex = logError $ "Unexpected Finish" :# ["exception" .= displayException ex]
library/Freckle/App/Bugsnag.hs view
@@ -22,8 +22,8 @@ import Freckle.App.Prelude -import Control.Concurrent (forkIO) import Control.Lens (Lens', view)+import Control.Monad.Catch (MonadMask) import Control.Monad.Reader (runReaderT) import Data.Bugsnag import Data.Bugsnag.Settings@@ -31,6 +31,7 @@ import Data.List (isInfixOf) import Database.PostgreSQL.Simple (SqlError (..)) import Database.PostgreSQL.Simple.Errors+import Freckle.App.Async (async) import qualified Freckle.App.Env as Env import Network.Bugsnag hiding (notifyBugsnag, notifyBugsnagWith) import qualified Network.Bugsnag as Bugsnag@@ -64,10 +65,11 @@ -- | Notify Bugsnag of an exception ----- The notification is made asynchronously via a simple @'forkIO'@. This is+-- The notification is made asynchronously via a simple @'async'@. This is -- best-effort and we don't care to keep track of the spawned threads. notifyBugsnag- :: ( MonadIO m+ :: ( MonadMask m+ , MonadUnliftIO m , MonadReader env m , HasBugsnagSettings env , Exception.Exception e@@ -78,7 +80,8 @@ -- | 'notifyBugsnag' with a 'BeforeNotify' notifyBugsnagWith- :: ( MonadIO m+ :: ( MonadMask m+ , MonadUnliftIO m , MonadReader env m , HasBugsnagSettings env , Exception.Exception e@@ -88,7 +91,7 @@ -> m () notifyBugsnagWith f ex = do settings <- view bugsnagSettingsL- void $ liftIO $ forkIO $ Bugsnag.notifyBugsnagWith f settings ex+ void $ async $ liftIO $ Bugsnag.notifyBugsnagWith f settings ex asSqlError :: SqlError -> BeforeNotify asSqlError err@SqlError {..} = toSqlGrouping <> toSqlException
library/Freckle/App/Kafka/Consumer.hs view
@@ -11,15 +11,13 @@ import Freckle.App.Prelude import Blammo.Logging-import Control.Concurrent-import qualified Control.Immortal as Immortal import Control.Lens (Lens', view)-import Control.Monad (forever) import Data.Aeson import qualified Data.List.NonEmpty as NE import qualified Data.Map.Strict as Map import qualified Data.Text as T import qualified Env+import Freckle.App.Async import Freckle.App.Env import Freckle.App.Kafka.Producer (envKafkaBrokerAddresses) import Kafka.Consumer hiding@@ -30,7 +28,7 @@ , subscription ) import qualified Kafka.Consumer as Kafka-import UnliftIO.Exception (bracket, displayException, throwIO)+import UnliftIO.Exception (bracket, throwIO) data KafkaConsumerConfig = KafkaConsumerConfig { kafkaConsumerConfigBrokerAddresses :: NonEmpty BrokerAddress@@ -145,7 +143,8 @@ TimeoutMilliseconds ms -> ms runConsumer- :: ( MonadUnliftIO m+ :: ( MonadMask m+ , MonadUnliftIO m , MonadReader env m , MonadLogger m , HasKafkaConsumer env@@ -154,22 +153,14 @@ => Timeout -> (a -> m ()) -> m ()-runConsumer pollTimeout onMessage = do- void $ Immortal.create $ \thread -> Immortal.onUnexpectedFinish thread handleException $ do- consumer <- view kafkaConsumerL- eMessage <-- pollMessage consumer $ Kafka.Timeout $ timeoutMs pollTimeout- case eMessage of- Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug "Polling timeout"- Left err -> logError $ "Error polling for message from Kafka" :# ["error" .= show err]- Right ConsumerRecord {..} -> for_ crValue $ \bs ->- case eitherDecodeStrict bs of- Left err -> logError $ "Could not decode message value" :# ["error" .= err]- Right a -> onMessage a- forever $ liftIO $ threadDelay maxBound- where- handleException = \case- Left ex ->- logError $- "Exception occurred in runConsumer" :# ["exception" .= displayException ex]- Right () -> pure ()+runConsumer pollTimeout onMessage = immortalCreateLogged $ do+ consumer <- view kafkaConsumerL+ eMessage <-+ pollMessage consumer $ Kafka.Timeout $ timeoutMs pollTimeout+ case eMessage of+ Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug "Polling timeout"+ Left err -> logError $ "Error polling for message from Kafka" :# ["error" .= show err]+ Right ConsumerRecord {..} -> for_ crValue $ \bs ->+ case eitherDecodeStrict bs of+ Left err -> logError $ "Could not decode message value" :# ["error" .= err]+ Right a -> onMessage a
library/Freckle/App/Kafka/Producer.hs view
@@ -22,9 +22,9 @@ import Data.Pool (Pool) import qualified Data.Pool as Pool import qualified Data.Text as T+import Freckle.App.Async (async) import qualified Freckle.App.Env as Env import Kafka.Producer-import UnliftIO.Async (async) import UnliftIO.Exception (throwString) import Yesod.Core.Lens import Yesod.Core.Types (HandlerData)@@ -101,12 +101,12 @@ throw err = throwString $ "Failed to open kafka producer: " <> show err produceKeyedOn- :: ( ToJSON value- , ToJSON key+ :: ( MonadUnliftIO m , MonadLogger m , MonadReader env m , HasKafkaProducerPool env- , MonadUnliftIO m+ , ToJSON key+ , ToJSON value ) => TopicName -> NonEmpty value@@ -139,12 +139,13 @@ } produceKeyedOnAsync- :: ( ToJSON value- , ToJSON key+ :: ( MonadMask m+ , MonadUnliftIO m , MonadLogger m , MonadReader env m , HasKafkaProducerPool env- , MonadUnliftIO m+ , ToJSON key+ , ToJSON value ) => TopicName -> NonEmpty value
package.yaml view
@@ -1,5 +1,5 @@ name: freckle-app-version: 1.9.3.0+version: 1.9.4.0 maintainer: Freckle Education category: Utils github: freckle/freckle-app