Blockfrost speedup

CHANGELOG.md

@@ -25,6 +25,17 @@ changes.
     broadcast of 5000 messages went from thousands of connections to 5, and got
     ~12% faster.
 
+- Reworked the Blockfrost chain backend to poll verifiable conditions instead
+  of sleeping fixed delays: transaction awaits check inclusion and output
+  visibility, the chain follower processes blocks in batches staying about one
+  block behind the tip, submission errors are reported immediately (API
+  rejections were previously reported as success), and HTTP 429 is retried
+  with capped exponential backoff. A full head lifecycle on preview drops from
+  over an hour to minutes, and the Blockfrost lifecycle test runs in nightly
+  CI again. `--blockfrost-retry-timeout` now bounds transaction awaits in
+  seconds; the ineffective `--blockfrost-query-timeout` option and
+  `query-timeout` config key were removed.
+
 - Changed `hydra-cluster/config/protocol-parameters.json` so that no layer 2
   UTxO can become impossible to fan out on layer 1: `maxTxSize` lowered to
   10250 (fanout carries ~5.8 kB of overhead; safe up to 10 parties),

hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs

@@ -6,7 +6,6 @@ import Hydra.Prelude
 
 import Blockfrost.Client (
   BlockfrostClientT,
-  runBlockfrost,
  )
 import Blockfrost.Client qualified as Blockfrost
 import Control.Concurrent.Class.MonadSTM (
@@ -30,6 +29,7 @@ import Hydra.Cardano.Api (
 import Hydra.Cardano.Api.Prelude (
   BlockHeader (..),
  )
+import Hydra.Chain.Blockfrost.Client (maxRateLimitRetries, rateLimitBackoff)
 import Hydra.ChainObserver.NodeClient (
   ChainObservation (..),
   ChainObserverLog (..),
@@ -49,19 +49,31 @@ data APIBlockfrostError
   | NotEnoughBlockConfirmations Blockfrost.BlockHash
   | MissingBlockNo Blockfrost.BlockHash
   | MissingNextBlockHash Blockfrost.BlockHash
+  | BlockfrostRateLimited
   deriving stock (Show)
   deriving anyclass (Exception)
 
+-- | Run a Blockfrost client action, retrying with capped exponential backoff
+-- when rate limited (HTTP 429). blockfrost-client does not expose the
+-- Retry-After header, so the delay is blind: 1s, 2s, 4s ... capped at 60s.
+-- Gives up after 'maxRateLimitRetries' and throws 'BlockfrostRateLimited'.
 runBlockfrostM ::
   (MonadIO m, MonadThrow m) =>
   Blockfrost.Project ->
   BlockfrostClientT IO a ->
   m a
-runBlockfrostM prj action = do
-  result <- liftIO $ runBlockfrost prj action
-  case result of
-    Left err -> throwIO (BlockfrostError $ show err)
-    Right val -> pure val
+runBlockfrostM prj action = go 0
+ where
+  go attempt = do
+    result <- liftIO $ Blockfrost.runBlockfrost prj action
+    case result of
+      Right val -> pure val
+      Left Blockfrost.BlockfrostUsageLimitReached
+        | attempt < maxRateLimitRetries -> do
+            liftIO $ threadDelay (rateLimitBackoff attempt)
+            go (attempt + 1)
+        | otherwise -> throwIO BlockfrostRateLimited
+      Left err -> throwIO $ BlockfrostError (show err)
 
 blockfrostClient ::
   Tracer IO ChainObserverLog ->
@@ -197,6 +209,7 @@ isRetryable (DecodeError _) = False
 isRetryable (NotEnoughBlockConfirmations _) = True
 isRetryable (MissingBlockNo _) = True
 isRetryable (MissingNextBlockHash _) = True
+isRetryable BlockfrostRateLimited = True
 
 toChainPoint :: Blockfrost.Block -> ChainPoint
 toChainPoint Blockfrost.Block{_blockSlot, _blockHash} =

hydra-cluster/src/CardanoNode.hs

@@ -261,13 +261,6 @@ withBlockfrostBackend _tracer stateDirectory action = do
   shelleyGenesis <- readFileBS >=> unsafeDecodeJson $ stateDirectory </> nodeShelleyGenesisFile args
   bfProjectPath <- findFileStartingAtDirectory 3 Backend.blockfrostProjectPath
   let opts = Options.Blockfrost defaultBlockfrostOptions{projectPath = bfProjectPath}
-  -- We need to make sure somehow that, before we start our blockfrost tests,
-  -- doing queries will give us updated information on some UTxO. There is no
-  -- way to definitely know if this information is correct since it might be
-  -- outdated. We just try to wait for sufficient amount of time before
-  -- starting another BF related test.
-  delay <- runBackend opts getQueryDelay
-  threadDelay $ realToFrac delay
   action (getShelleyGenesisBlockTime shelleyGenesis) opts
 
 -- | Find the given file in the current directory or its parents.

hydra-cluster/src/Hydra/Cluster/Faucet.hs

@@ -29,9 +29,9 @@ import Hydra.Chain.ScriptRegistry (
 import Hydra.Cluster.Fixture (Actor (Faucet))
 import Hydra.Cluster.Util (keysFor)
 import Hydra.Ledger.Cardano ()
-import Hydra.Options (ChainBackendOptions (..), defaultBFQueryTimeout)
+import Hydra.Options (ChainBackendOptions (..))
 import Hydra.Options qualified as Options
-import Hydra.Tx (balance, txId)
+import Hydra.Tx (balance)
 import Hydra.Tx.Crypto (getVerificationKey, signTx)
 import Hydra.Tx.Secret (Secret, mkSecret, withSecret)
 import System.Directory (doesFileExist)
@@ -55,7 +55,7 @@ data FaucetLog
 delayBF :: MonadDelay m => ChainBackendOptions -> m ()
 delayBF opts = do
   let delay = case opts of
-        Options.Blockfrost{} -> defaultBFQueryTimeout
+        Options.Blockfrost{} -> 30 :: Int -- backoff before retrying a failed BF faucet operation
         _ -> 1
   threadDelay $ fromIntegral delay
 
@@ -68,7 +68,6 @@ seedFromFaucet ::
   Tracer IO FaucetLog ->
   IO UTxO
 seedFromFaucet opts receivingVerificationKey val tracer = do
-  delayBF opts
   seedFromFaucetWithMinting opts receivingVerificationKey val tracer Nothing
 
 -- | Create a specially marked "seed" UTXO containing requested 'Value' by
@@ -145,7 +144,7 @@ seedFromFaucetBlockfrost options receivingVerificationKey lovelace = do
   let stakePools = Set.fromList (Blockfrost.toCardanoPoolId <$> stakePools')
   let systemStart = SystemStart $ posixSecondsToUTCTime systemStart'
   eraHistory <- Blockfrost.queryEraHistory
-  faucetUTxO <- Blockfrost.queryUTxO options networkId [changeAddress]
+  faucetUTxO <- Blockfrost.queryUTxO networkId [changeAddress]
   foundUTxO <- findUTxO faucetUTxO lovelace
   case buildTransactionWithPParams' pparams systemStart eraHistory stakePools (mkVkAddress networkId faucetVk) foundUTxO [] [theOutput] Nothing of
     Left e -> liftIO $ throwIO $ FaucetFailedToBuildTx{reason = e}
@@ -155,8 +154,8 @@ seedFromFaucetBlockfrost options receivingVerificationKey lovelace = do
       case eResult of
         Left err -> liftIO $ throwIO $ FaucetBlockfrostError{blockFrostError = show err}
         Right _ -> do
-          void $ Blockfrost.awaitUTxO networkId [changeAddress] (Hydra.Tx.txId signedTx) options
-          Blockfrost.awaitUTxO networkId [receivingAddress] (Hydra.Tx.txId signedTx) options
+          void $ Blockfrost.awaitUTxO networkId [changeAddress] signedTx options
+          Blockfrost.awaitUTxO networkId [receivingAddress] signedTx options
 
 findUTxO :: MonadIO m => UTxO.UTxO Era -> Lovelace -> m (UTxO.UTxO Era)
 findUTxO utxo lovelace' = do
@@ -186,7 +185,6 @@ returnFundsToFaucet ::
   Actor ->
   IO ()
 returnFundsToFaucet tracer opts sender = do
-  delayBF opts
   senderKeys <- keysFor sender
   void $ returnFundsToFaucet' tracer opts (snd senderKeys)
 
@@ -277,9 +275,7 @@ retryOnExceptions tracer opts action =
 publishHydraScriptsAs :: ChainBackendOptions -> Actor -> IO [TxId]
 publishHydraScriptsAs opts actor = do
   (_, sk) <- keysFor actor
-  txids <- runBackend opts $ publishHydraScripts (withSecret sk (mkSecret . CardanoSigningKey))
-  delayBF opts
-  pure txids
+  runBackend opts $ publishHydraScripts (withSecret sk (mkSecret . CardanoSigningKey))
 
 -- | Like 'publishHydraScriptsAs', but caches the resulting 'TxId's to a file
 -- in the given directory. On subsequent calls, the cached 'TxId's are validated

hydra-cluster/src/Hydra/Cluster/Scenarios.hs

@@ -2045,7 +2045,6 @@ refuelIfNeeded ::
   Coin ->
   IO ()
 refuelIfNeeded tracer opts actor amount = do
-  Faucet.delayBF opts
   (actorVk, _) <- keysFor actor
   existingUtxo <- runBackend opts $ queryUTxOFor QueryTip actorVk
   traceWith tracer $ StartingFunds{actor = actorName actor, utxo = existingUtxo}

hydra-cluster/src/HydraNode.hs

@@ -30,7 +30,7 @@ import Hydra.Logging (Tracer, Verbosity (..), traceWith)
 import Hydra.Network (Host (Host), NodeId (NodeId), WhichEtcd (SystemEtcd))
 import Hydra.Network qualified as Network
 import Hydra.Network.Etcd (peerPortToClientPort)
-import Hydra.Options (BlockfrostOptions (..), CardanoChainConfig (..), ChainBackendOptions (..), ChainConfig (..), DirectOptions (..), LedgerConfig (..), RunOptions (..), defaultBFQueryTimeout, defaultCardanoChainConfig, defaultDirectOptions, nodeSocket, toArgs)
+import Hydra.Options (BlockfrostOptions (..), CardanoChainConfig (..), ChainBackendOptions (..), ChainConfig (..), DirectOptions (..), LedgerConfig (..), RunOptions (..), defaultCardanoChainConfig, defaultDirectOptions, nodeSocket, toArgs)
 import Hydra.Tx (ConfirmedSnapshot)
 import Hydra.Tx.Crypto (HydraKey, getVerificationKey)
 import Hydra.Tx.Secret (Secret, withSecret)
@@ -101,7 +101,10 @@ output tag pairs = object $ ("tag" .= tag) : pairs
 setupBFDelay :: NominalDiffTime -> IO NominalDiffTime
 setupBFDelay d = do
   Prelude.getHydraNetwork >>= \case
-    Prelude.Blockfrost -> pure $ d * fromIntegral defaultBFQueryTimeout
+    -- The Blockfrost follower observes ~1 block behind tip plus one poll
+    -- interval, on a network with much longer block times than the devnet
+    -- timings most waits are written for.
+    Prelude.Blockfrost -> pure $ d * 3
     _backend -> pure d
 
 -- | Wait some time for a single API server output from each of given nodes.

hydra-cluster/test/Test/BlockfrostChainSpec.hs

@@ -9,7 +9,6 @@ import Cardano.Api.UTxO qualified as UTxO
 import Control.Concurrent.STM (takeTMVar)
 import Control.Concurrent.STM.TMVar (putTMVar)
 import Control.Exception (IOException)
-import Data.Time (secondsToNominalDiffTime)
 import Hydra.Cardano.Api (CardanoSigningKey (..), TxIn (..), TxIx (..), pattern TxOut, pattern TxOutDatumInline)
 import Hydra.Chain (
   Chain (Chain, postTx),
@@ -62,8 +61,6 @@ import Test.DirectChainSpec (
   observesInTimeSatisfying',
   waitMatch,
  )
-import Test.Hydra.Tx.Gen (genKeyPair)
-import Test.QuickCheck (generate)
 
 spec :: Spec
 spec = around (onlyWithBlockfrostProjectFile . showLogsOnFailure "BlockfrostChainSpec") $ do
@@ -89,44 +86,51 @@ spec = around (onlyWithBlockfrostProjectFile . showLogsOnFailure "BlockfrostChai
       failure $
         "Expected a published reference output to preserve its inline datum, but none did: " <> show utxo
 
-  -- Parked until #2753 makes the Blockfrost lifecycle fast enough to run in CI.
-  it "can open, close & fanout a Head using Blockfrost" $ \tracer -> do
-    pendingWith "Blockfrost tests should run only as part of smoke-tests because they are very slow"
+  -- NOTE: re-running within a minute of an aborted run can fail on submission with "all inputs are spent",
+  -- because the shared faucet's address index may not yet reflect the previous process's transactions;
+  -- wait a minute and re-run. (The retry hardening stays deferred; it's recorded in the plan.)
+  it "can open, close & fanout a Head using Blockfrost @requiresBlockfrost" $ \tracer -> do
     withTempDir "hydra-cluster" $ \tmp -> do
       (_, sk) <- keysFor Faucet
       prj <- Blockfrost.projectFromFile blockfrostProjectPath
       (aliceCardanoVk, _) <- keysFor Alice
-      (aliceExternalVk, _aliceExternalSk) <- generate genKeyPair
       let blockfrostOpts = defaultBlockfrostOptions{projectPath = blockfrostProjectPath}
       hydraScriptsTxId <- runBlockfrostBackend blockfrostOpts $ publishHydraScripts (withSecret sk (mkSecret . CardanoSigningKey))
 
       Blockfrost.Genesis
         { _genesisNetworkMagic
         , _genesisSystemStart
+        , _genesisSlotLength
+        , _genesisActiveSlotsCoefficient
         } <-
         Blockfrost.runBlockfrostM prj Blockfrost.queryGenesisParameters
 
+      let blockTime :: NominalDiffTime
+          blockTime = realToFrac _genesisSlotLength / realToFrac _genesisActiveSlotsCoefficient
+      -- Inclusion takes 1-2 blocks and the follower observes with ~1 block of
+      -- confirmation lag plus one poll interval; 6 block times gives margin.
+      let observationTimeout = 6 * blockTime
+
       -- Alice setup
       aliceChainConfig <- chainConfigFor' Alice tmp (Blockfrost blockfrostOpts) hydraScriptsTxId [] blockfrostcperiod (DepositPeriod 100) (DepositPeriod 100)
 
       withBlockfrostChainTest (contramap (FromBlockfrostChain "alice") tracer) aliceChainConfig alice $
         \aliceChain@CardanoChainTest{postTx} -> do
           _ <- Blockfrost.runBlockfrostM prj $ seedFromFaucetBlockfrost defaultBlockfrostOptions aliceCardanoVk 100_000_000
-          someUTxO <- Blockfrost.runBlockfrostM prj $ seedFromFaucetBlockfrost defaultBlockfrostOptions aliceExternalVk 7_000_000
           -- Scenario
           participants <- loadParticipants [Alice]
           let headParameters = HeadParameters blockfrostcperiod (DepositPeriod 100) [alice]
           postTx $ InitTx{participants, headParameters}
-          (headId, headSeed) <- observesInTimeSatisfying' aliceChain (secondsToNominalDiffTime $ fromIntegral $ queryTimeout defaultBlockfrostOptions) $ hasInitTxWith headParameters participants
+          (headId, headSeed) <- observesInTimeSatisfying' aliceChain observationTimeout $ hasInitTxWith headParameters participants
 
-          -- TODO: Deposit someUTxO
           let snapshotVersion = 0
-          let accumulator = Accumulator.buildFromUTxO someUTxO
+          let emptyUTxO :: UTxOType Tx = mempty
+          let accumulator = Accumulator.buildFromUTxO emptyUTxO
           let snapshot =
                 Snapshot
                   { headId
                   , number = 1
-                  , utxo = someUTxO
+                  , utxo = emptyUTxO
                   , confirmed = []
                   , utxoToCommit = Nothing
                   , utxoToDecommit = Nothing
@@ -157,7 +161,7 @@ spec = around (onlyWithBlockfrostProjectFile . showLogsOnFailure "BlockfrostChai
           let expectedUTxO =
                 (Snapshot.utxo snapshot <> fromMaybe mempty (Snapshot.utxoToCommit snapshot))
                   `withoutUTxO` fromMaybe mempty (Snapshot.utxoToDecommit snapshot)
-          observesInTimeSatisfying' aliceChain (secondsToNominalDiffTime $ fromIntegral $ queryTimeout defaultBlockfrostOptions) $ \case
+          observesInTimeSatisfying' aliceChain observationTimeout $ \case
             OnFanoutTx{headId = headId', fanoutUTxO}
               | headId' == headId ->
                   if UTxO.containsOutputs fanoutUTxO (UTxO.txOutputs expectedUTxO)