Limit parallel requests to Ogmios

packages.dhall

@@ -3,4 +3,52 @@ let upstream =
       https://raw.githubusercontent.com/mlabs-haskell/purescript-cardano-package-set/v3.1.0/packages.dhall
         sha256:0d8a7ca4e8ecfc8d1d795a989b76364caa9583d60e765c490cfa215a8824c246
 
-in upstream
+let additions =
+      { cardano-kupmios-provider =
+        { dependencies =
+          [ "aeson"
+          , "aff"
+          , "affjax"
+          , "arrays"
+          , "bifunctors"
+          , "bytearrays"
+          , "cardano-provider"
+          , "cardano-data-lite"
+          , "cardano-types"
+          , "concurrent-queues"
+          , "console"
+          , "control"
+          , "datetime"
+          , "effect"
+          , "either"
+          , "exceptions"
+          , "foldable-traversable"
+          , "foreign-object"
+          , "formatters"
+          , "http-methods"
+          , "integers"
+          , "js-bigints"
+          , "js-date"
+          , "maybe"
+          , "monad-logger"
+          , "newtype"
+          , "ordered-collections"
+          , "parallel"
+          , "partial"
+          , "prelude"
+          , "profunctor-lenses"
+          , "record"
+          , "strings"
+          , "stringutils"
+          , "tailrec"
+          , "transformers"
+          , "tuples"
+          , "uint"
+          , "untagged-union"
+          ]
+        , repo = "https://github.com/mlabs-haskell/purescript-cardano-kupmios-provider"
+        , version = "22044b695b6d3e43724c01ed5f255ece33dc4744" 
+        }
+      }
+
+in (upstream // additions)

spago.dhall

@@ -30,6 +30,7 @@
   , "cip30-typesafe"
   , "cip95"
   , "cip95-typesafe"
+  , "concurrent-queues"
   , "console"
   , "control"
   , "crypto"

src/Internal/Contract/Monad.purs

@@ -22,17 +22,16 @@ module Ctl.Internal.Contract.Monad
 
 import Prelude
 
-import Cardano.Blockfrost.Service
-  ( BlockfrostServiceM
-  , runBlockfrostServiceM
-  )
+import Cardano.Blockfrost.Service (BlockfrostServiceM, runBlockfrostServiceM)
 import Cardano.Blockfrost.Service as Blockfrost
-import Cardano.Kupmios.KupmiosM (KupmiosEnv, KupmiosM)
-import Cardano.Kupmios.Ogmios (getProtocolParameters, getSystemStartTime)
-import Cardano.Kupmios.Ogmios.Types
-  ( OgmiosDecodeError
-  , pprintOgmiosDecodeError
+import Cardano.Kupmios
+  ( KupmiosConfig
+  , KupmiosEnv
+  , KupmiosM
+  , initOgmiosRequestSemaphore
   )
+import Cardano.Kupmios.Ogmios (getProtocolParameters, getSystemStartTime)
+import Cardano.Kupmios.Ogmios.Types (OgmiosDecodeError, pprintOgmiosDecodeError)
 import Cardano.Provider.Error (ClientError)
 import Cardano.Provider.Type (Provider)
 import Cardano.Types (NetworkId(TestnetId, MainnetId), TransactionHash, UtxoMap)
@@ -79,7 +78,7 @@ import Data.Maybe (Maybe(Just, Nothing), fromMaybe)
 import Data.Newtype (class Newtype, unwrap)
 import Data.Set (Set)
 import Data.Set as Set
-import Data.Time.Duration (Milliseconds, Seconds)
+import Data.Time.Duration (Milliseconds(Milliseconds), Seconds)
 import Data.Traversable (for_, traverse)
 import Effect.Aff (Aff, ParAff, attempt, error, finally, supervise)
 import Effect.Aff.Class (class MonadAff, liftAff)
@@ -264,8 +263,10 @@ buildBackend _ = case _ of
   where
   buildCtlBackend :: CtlBackendParams -> Aff CtlBackend
   buildCtlBackend { ogmiosConfig, kupoConfig } = do
+    sem <- initOgmiosRequestSemaphore { maxParallelRequests: 5 }
     pure
       { ogmiosConfig
+      , ogmiosRequestSemaphore: Just sem
       , kupoConfig
       }
 
@@ -446,25 +447,32 @@ wrapKupmiosM qm = do
   liftAff $ runKupmiosM contractEnv ctlBackend qm
 
 runKupmiosM
-  :: forall (a :: Type) (rest :: Row Type)
-   . LogParams rest
+  :: forall (a :: Type) (r :: Row Type)
+   . LogParams r
   -> CtlBackend
   -> KupmiosM a
   -> Aff a
-runKupmiosM params ctlBackend =
-  flip runReaderT (mkKupmiosEnv params ctlBackend) <<< unwrap
-
-mkKupmiosEnv
-  :: forall (rest :: Row Type). LogParams rest -> CtlBackend -> KupmiosEnv
-mkKupmiosEnv params ctlBackend =
-  { config:
-      { ogmiosConfig: ctlBackend.ogmiosConfig
-      , kupoConfig: ctlBackend.kupoConfig
-      , logLevel: params.logLevel
-      , customLogger: params.customLogger
-      , suppressLogs: params.suppressLogs
-      }
-  }
+runKupmiosM params ctlBackend = flip runReaderT env <<< unwrap
+  where
+  env :: KupmiosEnv
+  env =
+    { config
+    , ogmiosRequestSemaphore: ctlBackend.ogmiosRequestSemaphore
+    }
+
+  config :: KupmiosConfig
+  config =
+    { ogmios:
+        { serverConfig: ctlBackend.ogmiosConfig
+        , requestSemaphoreCooldown: Just $ Milliseconds 300.0
+        }
+    , kupo:
+        { serverConfig: ctlBackend.kupoConfig
+        }
+    , logLevel: params.logLevel
+    , customLogger: params.customLogger
+    , suppressLogs: params.suppressLogs
+    }
 
 --------------------------------------------------------------------------------
 -- Helpers

src/Internal/Contract/ProviderBackend.purs

@@ -15,6 +15,8 @@ module Ctl.Internal.Contract.ProviderBackend
 import Prelude
 
 import Cardano.Blockfrost.BlockfrostBackend (BlockfrostBackend)
+import Cardano.Kupmios.KupmiosM (Semaphore) as Kupmios
+import Concurrent.Queue (Queue)
 import Ctl.Internal.ServerConfig (ServerConfig)
 import Data.Maybe (Maybe(Just, Nothing))
 import Data.Time.Duration (Seconds(Seconds))
@@ -29,6 +31,7 @@ data ProviderBackend
 
 type CtlBackend =
   { ogmiosConfig :: ServerConfig
+  , ogmiosRequestSemaphore :: Maybe Kupmios.Semaphore
   , kupoConfig :: ServerConfig
   }