freckle-app 1.12.0.1 → 1.13.0.0

raw patch · 9 files changed

+166/−48 lines, 9 filesdep +autodocodecdep +autodocodec-openapi3dep +openapi3PVP ok

version bump matches the API change (PVP)

Dependencies added: autodocodec, autodocodec-openapi3, openapi3

API changes (from Hackage documentation)

Files

• CHANGELOG.md +12/−1
• freckle-app.cabal +21/−16
• library/Freckle/App/Exception/MonadThrow.hs +13/−0
• library/Freckle/App/Exception/MonadUnliftIO.hs +13/−0
• library/Freckle/App/Json/Empty.hs +37/−0
• library/Freckle/App/Kafka/Consumer.hs +41/−26
• library/Freckle/App/Memcached/Client.hs +1/−1
• library/Freckle/App/Yesod.hs +22/−3
• package.yaml +6/−1

@@ -1,4 +1,15 @@-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.12.0.1...main)+## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.12.0.0...main)++## [v1.13.0.0](https://github.com/freckle/freckle-app/compare/v1.12.0.1...v1.13.0.0)++- Handle committing offsets more intelligently in `Kafka.Consumer` [breaking]++  The autocommit configuration and behavior was removed. Instead, we store and+  commit offsets explicitly such that at-least-once processing is ensured and+  duplicate processing (during deploy scaling) is minimized.++- Add `Freckle.App.Json.Empty` with the `Empty` type.+- Expose data constructors for `Freckle.App.Memcached.Client.MemcachedClient`.  ## [v1.12.0.1](https://github.com/freckle/freckle-app/compare/v1.12.0.0...v1.12.0.1) 

@@ -1,6 +1,6 @@ cabal-version:      1.18 name:               freckle-app-version:            1.12.0.1+version:            1.13.0.0 license:            MIT license-file:       LICENSE maintainer:         Freckle Education@@ -44,6 +44,7 @@         Freckle.App.Http         Freckle.App.Http.Paginate         Freckle.App.Http.Retry+        Freckle.App.Json.Empty         Freckle.App.Kafka         Freckle.App.Kafka.Consumer         Freckle.App.Kafka.Producer@@ -79,11 +80,11 @@     default-language:   Haskell2010     default-extensions:         BangPatterns DataKinds DeriveAnyClass DeriveFoldable DeriveFunctor-        DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies-        FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving-        LambdaCase MultiParamTypeClasses NoImplicitPrelude-        NoMonomorphismRestriction OverloadedStrings RankNTypes-        RecordWildCards ScopedTypeVariables StandaloneDeriving+        DeriveGeneric DeriveLift DeriveTraversable DerivingVia+        DerivingStrategies FlexibleContexts FlexibleInstances GADTs+        GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses+        NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings+        RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving         TypeApplications TypeFamilies      ghc-options:@@ -96,8 +97,11 @@         Blammo >=1.0.2.2,         Glob >=0.10.2,         MonadRandom >=0.5.3,+        QuickCheck >=2.14.2,         aeson >=1.5.6.0,         annotated-exception >=0.2.0.5,+        autodocodec >=0.2.2.0,+        autodocodec-openapi3 >=0.2.1.1,         aws-xray-client-persistent >=0.1.0.5,         aws-xray-client-wai >=0.1.0.2,         base >=4.14.3.0 && <5,@@ -143,6 +147,7 @@         mtl >=2.2.2,         network-uri >=2.6.4.1,         nonempty-containers >=0.3.4.4,+        openapi3 >=3.1.0,         path-pieces >=0.2.1,         persistent >=2.13.3.0,         persistent-postgresql >=2.13.4.0,@@ -190,11 +195,11 @@     default-language:   Haskell2010     default-extensions:         BangPatterns DataKinds DeriveAnyClass DeriveFoldable DeriveFunctor-        DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies-        FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving-        LambdaCase MultiParamTypeClasses NoImplicitPrelude-        NoMonomorphismRestriction OverloadedStrings RankNTypes-        RecordWildCards ScopedTypeVariables StandaloneDeriving+        DeriveGeneric DeriveLift DeriveTraversable DerivingVia+        DerivingStrategies FlexibleContexts FlexibleInstances GADTs+        GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses+        NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings+        RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving         TypeApplications TypeFamilies      ghc-options:@@ -237,11 +242,11 @@     default-language:   Haskell2010     default-extensions:         BangPatterns DataKinds DeriveAnyClass DeriveFoldable DeriveFunctor-        DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies-        FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving-        LambdaCase MultiParamTypeClasses NoImplicitPrelude-        NoMonomorphismRestriction OverloadedStrings RankNTypes-        RecordWildCards ScopedTypeVariables StandaloneDeriving+        DeriveGeneric DeriveLift DeriveTraversable DerivingVia+        DerivingStrategies FlexibleContexts FlexibleInstances GADTs+        GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses+        NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings+        RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving         TypeApplications TypeFamilies      ghc-options:

@@ -8,6 +8,7 @@   , catches   , try   , tryJust+  , withException   , checkpoint   , checkpointMany   , checkpointCallStack@@ -23,6 +24,7 @@  import Control.Applicative (pure) import Control.Exception.Annotated (checkpoint, checkpointMany)+import Control.Monad (void) import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow) import Data.Either (Either (..)) import Data.Function (($), (.))@@ -103,6 +105,17 @@ tryJust test action =   withFrozenCallStack $ Annotated.catch (Right <$> action) $ \e ->     maybe (Control.Monad.Catch.throwM e) (pure . Left) (test e)++withException+  :: forall e a m b+   . (Exception e, MonadCatch m, HasCallStack)+  => m a+  -> (e -> m b)+  -> m a+withException action onException =+  withFrozenCallStack $ Annotated.catch action $ \e -> do+    void $ onException e+    Control.Monad.Catch.throwM e  -- | When dealing with a library that does not use 'AnnotatedException', --   apply this function to augment its exceptions with call stacks.

@@ -8,6 +8,7 @@   , catches   , try   , tryJust+  , withException   , checkpoint   , checkpointMany   , checkpointCallStack@@ -23,6 +24,7 @@  import Control.Applicative (pure) import Control.Exception.Annotated.UnliftIO (checkpoint, checkpointMany)+import Control.Monad (void) import Data.Either (Either (..)) import Data.Function (($), (.)) import Data.Functor (fmap, (<$>))@@ -104,6 +106,17 @@ tryJust test action =   withFrozenCallStack $ Annotated.catch (Right <$> action) $ \e ->     maybe (UnliftIO.Exception.throwIO e) (pure . Left) (test e)++withException+  :: forall e a m b+   . (Exception e, MonadUnliftIO m, HasCallStack)+  => m a+  -> (e -> m b)+  -> m a+withException action onException =+  withFrozenCallStack $ Annotated.catch action $ \e -> do+    void $ onException e+    UnliftIO.Exception.throwIO e  -- | When dealing with a library that does not use 'AnnotatedException', --   apply this function to augment its exceptions with call stacks.

@@ -0,0 +1,37 @@+-- | Aeson represents+module Freckle.App.Json.Empty+  ( Empty (..)+  ) where++import Freckle.App.Prelude++import Autodocodec (Autodocodec (..), HasCodec (..), HasObjectCodec (..), object)+import Autodocodec.OpenAPI ()+import Data.Aeson (FromJSON, ToJSON)+import Data.OpenApi (ToSchema (..))+import Test.QuickCheck (Arbitrary (..))++-- | A unit value encoded as an empty JSON object+--+-- Useful as the response body of a POST request when the server doesn't+-- need to return anything.+--+-- (One would expect to be able to use () for this, but Aeson encodes unit+-- as an empty list, not as an object.)+data Empty = Empty+  deriving (ToJSON, FromJSON, ToSchema) via (Autodocodec Empty)++instance Arbitrary Empty where+  arbitrary = pure Empty++instance HasCodec Empty where+  codec = object "Empty" objectCodec++instance HasObjectCodec Empty where+  objectCodec = pure Empty++instance Semigroup Empty where+  _ <> _ = Empty++instance Monoid Empty where+  mempty = Empty

@@ -20,6 +20,10 @@ import qualified Env import Freckle.App.Async import Freckle.App.Env+import Freckle.App.Exception+  ( AnnotatedException (..)+  , annotatedExceptionMessageFrom+  ) import Freckle.App.Kafka.Producer (envKafkaBrokerAddresses) import Freckle.App.OpenTelemetry import Kafka.Consumer hiding@@ -48,11 +52,6 @@   -- ^ The offset reset parameter used when there is no initial offset in Kafka.   --   -- This is the `auto.offset.reset` Kafka consumer configuration property.-  , kafkaConsumerConfigAutoCommitInterval :: Millis-  -- ^ The interval that offsets are auto-committed to Kafka.-  ---  -- This sets the `auto.commit.interval.ms` and `enable.auto.commit` Kafka-  -- consumer configuration properties.   , kafkaConsumerConfigExtraSubscriptionProps :: Map Text Text   -- ^ Extra properties used to configure the Kafka consumer.   }@@ -92,10 +91,6 @@   consumerGroupId <- Env.var Env.nonempty "KAFKA_CONSUMER_GROUP_ID" mempty   kafkaTopic <- envKafkaTopic   kafkaOffsetReset <- envKafkaOffsetReset-  kafkaAutoOffsetInterval <--    fromIntegral . timeoutMs <$$> Env.var timeout "KAFKA_AUTO_COMMIT_INTERVAL" $-      Env.def $-        TimeoutMilliseconds 5000   kafkaExtraProps <-     Env.var       (fmap Map.fromList . keyValues)@@ -107,7 +102,6 @@       consumerGroupId       kafkaTopic       kafkaOffsetReset-      kafkaAutoOffsetInterval       kafkaExtraProps  class HasKafkaConsumer env where@@ -117,7 +111,8 @@ consumerProps KafkaConsumerConfig {..} =   brokersList brokers     <> groupId kafkaConsumerConfigGroupId-    <> autoCommit kafkaConsumerConfigAutoCommitInterval+    <> noAutoCommit -- we handle offsets+    <> noAutoOffsetStore     <> logLevel KafkaLogInfo  where   brokers = NE.toList kafkaConsumerConfigBrokerAddresses@@ -172,31 +167,51 @@   -> (a -> m ())   -> m () runConsumer pollTimeout onMessage =-  withTraceIdContext $ immortalCreateLogged $ do+  withTraceIdContext $ immortalCreate onFinish $ do     consumer <- view kafkaConsumerL      flip catches handlers $ inSpan "kafka.consumer" consumerSpanArguments $ do       mRecord <- fromKafkaError =<< pollMessage consumer kTimeout -      for_ (crValue =<< mRecord) $ \bs -> do-        a <--          inSpan "kafka.consumer.message.decode" defaultSpanArguments $-            either (throwM . KafkaMessageDecodeError bs) pure $-              eitherDecodeStrict bs-        inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a+      for_ mRecord $ \r -> do+        for_ (crValue r) $ \bs -> do+          a <-+            inSpan "kafka.consumer.message.decode" defaultSpanArguments $+              either (throwM . KafkaMessageDecodeError bs) pure $+                eitherDecodeStrict bs+          inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a++        inSpan "kafka.consumer.message.commit" defaultSpanArguments $ do+          -- Store the offset of this record, then do a best-offort commit to+          -- the broker. If this fails and we crash or shutdown, the commit in+          -- the onFinish handler will pick it up.+          logExMay "Unable to store offset" $ storeOffsetMessage consumer r+          void $ commitAllOffsets OffsetCommitAsync consumer  where   kTimeout = Kafka.Timeout $ timeoutMs pollTimeout    handlers =-    [ ExceptionHandler $ \err ->-        logError $-          "Error polling for message from Kafka"-            :# ["error" .= displayException @KafkaError err]-    , ExceptionHandler $ \err ->-        logError $-          "Could not decode message value"-            :# ["error" .= displayException @KafkaMessageDecodeError err]+    [ ExceptionHandler $+        logErrorNS "kafka"+          . annotatedExceptionMessageFrom @KafkaError+            (const "Error polling for message from Kafka")+    , ExceptionHandler $+        logErrorNS "kafka"+          . annotatedExceptionMessageFrom @KafkaMessageDecodeError+            (const "Could not decode message value")     ]++  onFinish finishResult = do+    consumer <- view kafkaConsumerL+    logExMay "Unable to commit offsets" $ commitAllOffsets OffsetCommit consumer+    either (logEx "Unexpected finish") pure finishResult++  logExMay msg f = maybe (pure ()) (logEx msg) =<< f++  logEx msg =+    logErrorNS "kafka"+      . annotatedExceptionMessageFrom (const msg)+      . AnnotatedException []  fromKafkaError   :: (MonadIO m, MonadLogger m, HasCallStack)

@@ -1,5 +1,5 @@ module Freckle.App.Memcached.Client-  ( MemcachedClient+  ( MemcachedClient (..)   , newMemcachedClient   , withMemcachedClient   , memcachedClientDisabled

@@ -2,17 +2,25 @@ module Freckle.App.Yesod   ( respondQueryCanceled   , respondQueryCanceledHeaders+  , logExceptionsMiddleware   ) where  import Freckle.App.Prelude  import Blammo.Logging import Database.PostgreSQL.Simple (SqlError (..))+import Freckle.App.Exception+  ( AnnotatedException (..)+  , annotatedExceptionMessageFrom+  , fromException+  , withException+  ) import Freckle.App.Stats (HasStatsClient) import qualified Freckle.App.Stats as Stats import Network.HTTP.Types (ResponseHeaders, status503) import qualified Network.Wai as W import Yesod.Core.Handler (HandlerFor, sendWaiResponse)+import Yesod.Core.Types (HandlerContents)  -- | Catch 'SqlError' when queries are canceled due to timeout and respond 503 --@@ -29,9 +37,20 @@   -> HandlerFor site res respondQueryCanceledHeaders headers handler =   catchJust queryCanceled handler $ \ex -> do-    logError $ "Query canceled" :# ["exception" .= displayException ex]+    logErrorNS "yesod" $ annotatedExceptionMessageFrom (const "Query canceled") ex     Stats.increment "query_canceled"     sendWaiResponse $ W.responseLBS status503 headers "Query canceled" -queryCanceled :: SqlError -> Maybe SqlError-queryCanceled ex = ex <$ guard (sqlState ex == "57014")+logExceptionsMiddleware :: (MonadUnliftIO m, MonadLogger m) => m a -> m a+logExceptionsMiddleware f =+  f `withException` \ex ->+    unless (isHandlerContents ex) $+      logErrorNS "yesod" $+        annotatedExceptionMessageFrom (const "Handler exception") ex++isHandlerContents :: AnnotatedException SomeException -> Bool+isHandlerContents = isJust . fromException @HandlerContents . exception++queryCanceled+  :: AnnotatedException SqlError -> Maybe (AnnotatedException SqlError)+queryCanceled ex = ex <$ guard (sqlState (exception ex) == "57014")

@@ -1,5 +1,5 @@ name: freckle-app-version: 1.12.0.1+version: 1.13.0.0 maintainer: Freckle Education category: Utils github: freckle/freckle-app@@ -48,6 +48,7 @@   - DeriveGeneric   - DeriveLift   - DeriveTraversable+  - DerivingVia   - DerivingStrategies   - FlexibleContexts   - FlexibleInstances@@ -71,6 +72,9 @@     - Blammo     - Glob     - MonadRandom+    - QuickCheck+    - autodocodec+    - autodocodec-openapi3     - aeson     - annotated-exception     - aws-xray-client-persistent@@ -117,6 +121,7 @@     - mtl     - network-uri     - nonempty-containers+    - openapi3     - path-pieces     - persistent     - persistent-postgresql