diff --git a/CHANGELOG.md b/CHANGELOG.md index 1006a93844c..3529f05029c 100644 --- a/CHANGELOG.md +++ b/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), diff --git a/hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs b/hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs index 753db3d75d3..626004bf526 100644 --- a/hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs +++ b/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} = diff --git a/hydra-cluster/src/CardanoNode.hs b/hydra-cluster/src/CardanoNode.hs index 05223f879bb..40d5ba46fb6 100644 --- a/hydra-cluster/src/CardanoNode.hs +++ b/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. diff --git a/hydra-cluster/src/Hydra/Cluster/Faucet.hs b/hydra-cluster/src/Hydra/Cluster/Faucet.hs index 08a91492d05..b88323d0afa 100644 --- a/hydra-cluster/src/Hydra/Cluster/Faucet.hs +++ b/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 diff --git a/hydra-cluster/src/Hydra/Cluster/Scenarios.hs b/hydra-cluster/src/Hydra/Cluster/Scenarios.hs index 86accc7c8bb..7cea4ec6209 100644 --- a/hydra-cluster/src/Hydra/Cluster/Scenarios.hs +++ b/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} diff --git a/hydra-cluster/src/HydraNode.hs b/hydra-cluster/src/HydraNode.hs index 09162f5f1cf..8c2c25e0b2f 100644 --- a/hydra-cluster/src/HydraNode.hs +++ b/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. diff --git a/hydra-cluster/test/Test/BlockfrostChainSpec.hs b/hydra-cluster/test/Test/BlockfrostChainSpec.hs index dd08fb1f057..6f401cdb96a 100644 --- a/hydra-cluster/test/Test/BlockfrostChainSpec.hs +++ b/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) diff --git a/hydra-node/golden/RunOptions.json b/hydra-node/golden/RunOptions.json index a6bcd94ec0f..31a86359331 100644 --- a/hydra-node/golden/RunOptions.json +++ b/hydra-node/golden/RunOptions.json @@ -1,159 +1,166 @@ { "samples": [ { - "advertise": { - "hostname": "0.0.64.165", - "port": 188 - }, + "advertise": null, "apiHost": { - "ipv4": "0.0.2.73", + "ipv4": "0.0.82.151", "tag": "IPv4" }, - "apiPort": 322, - "apiTransactionTimeout": 19165, + "apiPort": 16485, + "apiTransactionTimeout": 54284, "chainConfig": { "cardanoSigningKey": "b.sk", "cardanoVerificationKeys": [ - "a.vk", - "a.vk", - "b.vk", - "c/a.vk" + "c/a.vk", + "a.vk" ], "chainBackendOptions": { "contents": { - "networkId": { - "magic": 42, - "tag": "Testnet" - }, - "nodeSocket": "node.socket" + "projectPath": "blockfrost-project.txt", + "retryTimeout": 300 }, - "tag": "Direct" + "tag": "Blockfrost" }, - "contestationPeriod": 604800, - "depositActivation": 53250, - "depositPeriod": 4126, + "contestationPeriod": 43200, + "depositActivation": 12339, + "depositPeriod": 9820, "hydraScriptsTxId": [ - "0300080601000107040003010105000702010004030100010503050600010306" + "0401040706040305050306010402060200050103010506010408020603040703", + "0401050503070307060800040204020706080003050300080102040804060305", + "0700080001020201060601030604040000080102000108000801040305020501", + "0701080105060206020605040108040507020002060201050806000500080505" ], "startChainFrom": null, "tag": "CardanoChainConfig", - "unsyncedPeriod": 82061 + "unsyncedPeriod": 74362 }, - "hydraSigningKey": "a/b/c/a/c/b.sk", + "hydraSigningKey": "a/a/a/c/c/c.sk", "hydraVerificationKeys": [ - "a/a/a.vk", - "a/b/a.vk", - "a.vk", - "a/a.vk" + "c/a.vk", + "b.vk" ], "ledgerConfig": { - "cardanoLedgerProtocolParametersFile": "c/a.json" + "cardanoLedgerProtocolParametersFile": "c/c/c/b/a.json" }, "listen": { - "hostname": "0.0.56.115", - "port": 23436 + "hostname": "0.0.88.75", + "port": 15797 }, - "monitoringPort": null, - "nodeId": "inwstimjzgtbgrpryky", + "monitoringPort": 30150, + "nodeId": "dqwz", "peers": [ { - "hostname": "0.0.0.7", - "port": 8 + "hostname": "0.0.0.6", + "port": 2 }, { - "hostname": "0.0.0.4", + "hostname": "0.0.0.5", "port": 2 }, { - "hostname": "0.0.0.8", - "port": 1 + "hostname": "0.0.0.4", + "port": 8 }, { - "hostname": "0.0.0.6", - "port": 2 + "hostname": "0.0.0.3", + "port": 1 } ], - "persistenceDir": "a/b/a/b/b/b", + "persistenceDir": "c", "persistenceRotateAfter": null, - "tlsCertPath": "a/b.pem", - "tlsKeyPath": "b/b/a.key", + "tlsCertPath": "c/b.pem", + "tlsKeyPath": "b/b.key", "verbosity": { - "tag": "Quiet" + "contents": "HydraNode", + "tag": "Verbose" }, - "whichEtcd": "SystemEtcd" + "whichEtcd": "EmbeddedEtcd" }, { "advertise": { - "hostname": "0.0.95.141", - "port": 8321 + "hostname": "0.0.105.152", + "port": 15129 }, "apiHost": { - "ipv4": "0.0.73.89", + "ipv4": "0.0.61.119", "tag": "IPv4" }, - "apiPort": 18849, - "apiTransactionTimeout": 1680, + "apiPort": 25761, + "apiTransactionTimeout": 78957, "chainConfig": { - "cardanoSigningKey": "b/c/a/b/b.sk", + "cardanoSigningKey": "a.sk", "cardanoVerificationKeys": [ - "c/c.vk", - "b.vk", - "a/b/c.vk" + "b/c/c.vk", + "c.vk", + "a/b/c.vk", + "b/a/c.vk", + "b/c/a.vk" ], "chainBackendOptions": { "contents": { - "projectPath": "blockfrost-project.txt", - "queryTimeout": 30, - "retryTimeout": 300 + "networkId": { + "magic": 42, + "tag": "Testnet" + }, + "nodeSocket": "node.socket" }, - "tag": "Blockfrost" + "tag": "Direct" }, - "contestationPeriod": 43200, - "depositActivation": 37366, - "depositPeriod": 70760, + "contestationPeriod": 604800, + "depositActivation": 78927, + "depositPeriod": 26387, "hydraScriptsTxId": [ - "0506030403020407070702060705040106050001030304020204030608050303", - "0108080401030806040702010803000105040706030308030107080805010303", - "0203040605000704070606070807010702020304000800040407030301020308", - "0401080205020306060103050708040102060107060102020607060401070803", - "0804000207080400020402080407020604040604050301020806050701050508", - "0104010400070407060408030203020707050802030702040506010601080101" + "0502010105070806040806000404020204010608000207020602050806080108", + "0507040005040508010104070304050405070000030501000005000703080806" ], - "startChainFrom": { - "blockHash": "b6712ec5372966b758cf92ff25d431654748ba7039b9065d14a3491224b48504", - "slot": 1859643, - "tag": "ChainPoint" - }, + "startChainFrom": null, "tag": "CardanoChainConfig", - "unsyncedPeriod": 35315 + "unsyncedPeriod": 68344 }, - "hydraSigningKey": "a/b/c/a.sk", + "hydraSigningKey": "c/c/c/a.sk", "hydraVerificationKeys": [ - "a/b.vk" + "b.vk", + "a/b/c.vk", + "a/a/c.vk", + "b/b/a.vk", + "a/c.vk", + "b/b.vk" ], "ledgerConfig": { - "cardanoLedgerProtocolParametersFile": "c/a/c/a.json" + "cardanoLedgerProtocolParametersFile": "a/c.json" }, "listen": { - "hostname": "0.0.102.182", - "port": 631 + "hostname": "0.0.57.217", + "port": 10710 }, - "monitoringPort": 27903, - "nodeId": "xrggjkhcynmqmkbrvrveifsgg", + "monitoringPort": null, + "nodeId": "pubvcfeoxelmmzuruxwodgz", "peers": [ + { + "hostname": "0.0.0.0", + "port": 5 + }, + { + "hostname": "0.0.0.1", + "port": 4 + }, + { + "hostname": "0.0.0.1", + "port": 0 + }, { "hostname": "0.0.0.4", "port": 7 }, { - "hostname": "0.0.0.3", - "port": 0 + "hostname": "0.0.0.5", + "port": 8 } ], - "persistenceDir": "a", - "persistenceRotateAfter": 16629, - "tlsCertPath": null, - "tlsKeyPath": "b/c/c/c/b/b.key", + "persistenceDir": "c/b/b/a/c/a", + "persistenceRotateAfter": null, + "tlsCertPath": "c/a/b/a/c/a.pem", + "tlsKeyPath": null, "verbosity": { "contents": "HydraNode", "tag": "Verbose" @@ -163,234 +170,175 @@ { "advertise": null, "apiHost": { - "ipv4": "0.0.48.42", + "ipv4": "0.0.91.111", "tag": "IPv4" }, - "apiPort": 20896, - "apiTransactionTimeout": 63349, + "apiPort": 22342, + "apiTransactionTimeout": 3456, "chainConfig": { - "cardanoSigningKey": "c/c/a.sk", - "cardanoVerificationKeys": [ - "a.vk", - "b.vk", - "b/c.vk", - "c/c/a.vk", - "a/a.vk" - ], - "chainBackendOptions": { - "contents": { - "networkId": { - "magic": 42, - "tag": "Testnet" - }, - "nodeSocket": "node.socket" - }, - "tag": "Direct" - }, - "contestationPeriod": 86400, - "depositActivation": 46021, - "depositPeriod": 13158, - "hydraScriptsTxId": [ - "0001050006080002030808040400080201000003030404010206080203070200" - ], - "startChainFrom": null, - "tag": "CardanoChainConfig", - "unsyncedPeriod": 10135 + "initialUTxOFile": "c/a/c.json", + "ledgerGenesisFile": null, + "offlineHeadSeed": "6862b57caef73c2d422b1f15cc6df206", + "tag": "OfflineChainConfig" }, - "hydraSigningKey": "c.sk", + "hydraSigningKey": "b/a/c/b.sk", "hydraVerificationKeys": [ - "a/a/a.vk", - "c/b/c.vk", - "c/c.vk", - "a.vk", - "b.vk", "b.vk" ], "ledgerConfig": { - "cardanoLedgerProtocolParametersFile": "b/a/a.json" + "cardanoLedgerProtocolParametersFile": "c/b/c/c/a.json" }, "listen": { - "hostname": "0.0.84.205", - "port": 21715 + "hostname": "0.0.36.136", + "port": 22676 }, - "monitoringPort": 11448, - "nodeId": "gikyesasxrnoyth", + "monitoringPort": null, + "nodeId": "mhuuoclfhrgjgobgu", "peers": [ { "hostname": "0.0.0.8", - "port": 5 + "port": 7 + }, + { + "hostname": "0.0.0.7", + "port": 3 }, { "hostname": "0.0.0.0", "port": 0 + }, + { + "hostname": "0.0.0.0", + "port": 6 + }, + { + "hostname": "0.0.0.6", + "port": 6 } ], - "persistenceDir": "c/b/a/c/a", + "persistenceDir": "b/a/b/a/b/b", "persistenceRotateAfter": null, "tlsCertPath": null, "tlsKeyPath": null, "verbosity": { - "contents": "HydraNode", - "tag": "Verbose" + "tag": "Quiet" }, "whichEtcd": "EmbeddedEtcd" }, { "advertise": { - "hostname": "0.0.122.222", - "port": 28105 + "hostname": "0.0.45.152", + "port": 19033 }, "apiHost": { - "ipv4": "0.0.27.242", + "ipv4": "0.0.23.217", "tag": "IPv4" }, - "apiPort": 10332, - "apiTransactionTimeout": 10840, + "apiPort": 7928, + "apiTransactionTimeout": 71086, "chainConfig": { - "cardanoSigningKey": "a/b/b/a/c/c.sk", - "cardanoVerificationKeys": [], - "chainBackendOptions": { - "contents": { - "projectPath": "blockfrost-project.txt", - "queryTimeout": 30, - "retryTimeout": 300 - }, - "tag": "Blockfrost" - }, - "contestationPeriod": 35649, - "depositActivation": 8176, - "depositPeriod": 10014, - "hydraScriptsTxId": [ - "0403050108080007020304000507050303060407010603050102010307080001", - "0000080000010404040503060607080003030105010603040508050107040804" - ], - "startChainFrom": { - "blockHash": "f38d9c0f863251b9c3ac29282c8f3675e72617d24038fdbb3265982e7b5e368c", - "slot": 8951344, - "tag": "ChainPoint" - }, - "tag": "CardanoChainConfig", - "unsyncedPeriod": 32654 + "initialUTxOFile": "b/b/c/c.json", + "ledgerGenesisFile": "c.json", + "offlineHeadSeed": "9e2b67d4cd8af225149d2529a0d797fd", + "tag": "OfflineChainConfig" }, - "hydraSigningKey": "b/c.sk", + "hydraSigningKey": "a/b/a.sk", "hydraVerificationKeys": [], "ledgerConfig": { - "cardanoLedgerProtocolParametersFile": "b/c/c/b/a.json" + "cardanoLedgerProtocolParametersFile": "c/a.json" }, "listen": { - "hostname": "0.0.9.134", - "port": 17520 + "hostname": "0.0.73.164", + "port": 14180 }, - "monitoringPort": 15487, - "nodeId": "xawuduxodfvu", + "monitoringPort": null, + "nodeId": "clcpxmwuqtfbdfjlynwavzsyqw", "peers": [ { - "hostname": "0.0.0.1", - "port": 4 + "hostname": "0.0.0.3", + "port": 2 }, { "hostname": "0.0.0.1", - "port": 4 - }, - { - "hostname": "0.0.0.2", - "port": 1 - }, - { - "hostname": "0.0.0.2", - "port": 8 + "port": 6 }, { - "hostname": "0.0.0.5", - "port": 8 + "hostname": "0.0.0.1", + "port": 3 } ], - "persistenceDir": "a/c/a/b", + "persistenceDir": "c/a/b/a", "persistenceRotateAfter": null, - "tlsCertPath": "c/b/c/c/a/b.pem", - "tlsKeyPath": "a/b.key", + "tlsCertPath": "c/b/c/c/c.pem", + "tlsKeyPath": "b/c/b/b/b.key", "verbosity": { "tag": "Quiet" }, - "whichEtcd": "EmbeddedEtcd" + "whichEtcd": "SystemEtcd" }, { "advertise": { - "hostname": "0.0.59.201", - "port": 12401 + "hostname": "0.0.60.49", + "port": 11249 }, "apiHost": { - "ipv4": "0.0.114.100", + "ipv4": "0.0.111.49", "tag": "IPv4" }, - "apiPort": 8958, - "apiTransactionTimeout": 67992, + "apiPort": 18616, + "apiTransactionTimeout": 67402, "chainConfig": { - "cardanoSigningKey": "b/c.sk", - "cardanoVerificationKeys": [ - "c/c/b.vk", - "b/c.vk", - "c/b/c.vk", - "b/a/a.vk" - ], - "chainBackendOptions": { - "contents": { - "projectPath": "blockfrost-project.txt", - "queryTimeout": 30, - "retryTimeout": 300 - }, - "tag": "Blockfrost" - }, - "contestationPeriod": 604800, - "depositActivation": 51706, - "depositPeriod": 3450, - "hydraScriptsTxId": [ - "0207000303050001080107060605010804070408030708070705060401030504", - "0201010601080805020304050202030204080005000001020103060108060201", - "0404040508080404010404060805030006020303000603020601020803060206", - "0805070700070704080206030202010300050602080204000601060305070405" - ], - "startChainFrom": { - "blockHash": "197d584d4cf360cade2318837c4870e94f75678fe6e07713fd8d59ad4b557f92", - "slot": 14915244, - "tag": "ChainPoint" - }, - "tag": "CardanoChainConfig", - "unsyncedPeriod": 85326 + "initialUTxOFile": "a/c/a/a/b.json", + "ledgerGenesisFile": null, + "offlineHeadSeed": "34650ee65b90de041a6a69a25bf7fe5d", + "tag": "OfflineChainConfig" }, - "hydraSigningKey": "c/c/c.sk", + "hydraSigningKey": "b/a/a.sk", "hydraVerificationKeys": [ - "c.vk", - "c.vk", "a.vk", - "a/b/c.vk", - "a/c/a.vk" + "a/a.vk", + "a.vk" ], "ledgerConfig": { - "cardanoLedgerProtocolParametersFile": "b.json" + "cardanoLedgerProtocolParametersFile": "a/c/c/b/b.json" }, "listen": { - "hostname": "0.0.91.10", - "port": 25097 + "hostname": "0.0.103.247", + "port": 5887 }, - "monitoringPort": 2546, - "nodeId": "jxvylesghffkctvhuiqfpsxw", + "monitoringPort": null, + "nodeId": "jbpvdtvtxdpimez", "peers": [ { - "hostname": "0.0.0.3", - "port": 6 + "hostname": "0.0.0.8", + "port": 1 + }, + { + "hostname": "0.0.0.8", + "port": 3 + }, + { + "hostname": "0.0.0.8", + "port": 1 + }, + { + "hostname": "0.0.0.7", + "port": 5 + }, + { + "hostname": "0.0.0.4", + "port": 0 } ], - "persistenceDir": "a", - "persistenceRotateAfter": 6302, - "tlsCertPath": "a/b.pem", + "persistenceDir": "c/a/c/b", + "persistenceRotateAfter": 26728, + "tlsCertPath": "a/a/a/b/b.pem", "tlsKeyPath": null, "verbosity": { - "contents": "HydraNode", - "tag": "Verbose" + "tag": "Quiet" }, "whichEtcd": "SystemEtcd" } ], - "seed": -1251614046 + "seed": 613889959 } \ No newline at end of file diff --git a/hydra-node/hydra-node.cabal b/hydra-node/hydra-node.cabal index 7aed86aeecf..f39ff0743a5 100644 --- a/hydra-node/hydra-node.cabal +++ b/hydra-node/hydra-node.cabal @@ -485,6 +485,7 @@ test-suite tests , regex-tdfa , req , resourcet + , retry , silently , sop-extras , sqlite-simple diff --git a/hydra-node/json-schemas/api.yaml b/hydra-node/json-schemas/api.yaml index 234d2776e84..c0ef2d0f345 100644 --- a/hydra-node/json-schemas/api.yaml +++ b/hydra-node/json-schemas/api.yaml @@ -4148,7 +4148,6 @@ components: required: - mode - project-path - - query-timeout - retry-timeout properties: mode: @@ -4157,9 +4156,6 @@ components: project-path: type: string description: Path to the Blockfrost project ID file. - query-timeout: - type: number - description: Timeout in seconds for Blockfrost API queries. retry-timeout: type: number description: Total timeout in seconds for retrying failed Blockfrost queries. diff --git a/hydra-node/src/Hydra/Chain/Backend.hs b/hydra-node/src/Hydra/Chain/Backend.hs index a36816836d1..7461ec7b0e3 100644 --- a/hydra-node/src/Hydra/Chain/Backend.hs +++ b/hydra-node/src/Hydra/Chain/Backend.hs @@ -28,9 +28,6 @@ class ChainBackend m where awaitTransaction :: Tx -> VerificationKey PaymentKey -> m UTxO getBlockTime :: m NominalDiffTime - -- | Get the delay to use between backend queries for rate limiting - getQueryDelay :: m NominalDiffTime - buildTransaction :: ChainBackend m => MonadIO m => diff --git a/hydra-node/src/Hydra/Chain/Blockfrost.hs b/hydra-node/src/Hydra/Chain/Blockfrost.hs index baeb7d394b8..343ad2cf333 100644 --- a/hydra-node/src/Hydra/Chain/Blockfrost.hs +++ b/hydra-node/src/Hydra/Chain/Blockfrost.hs @@ -7,7 +7,7 @@ import Control.Concurrent.Class.MonadSTM (putTMVar, readTQueue, readTVarIO, take import Control.Exception (IOException) import Control.Monad.Catch (Handler (Handler)) import Control.Monad.Catch qualified as Catch -import Control.Retry (RetryPolicyM, RetryStatus (..), constantDelay, fullJitterBackoff, limitRetries, recovering, retrying) +import Control.Retry (RetryPolicyM, RetryStatus (..), capDelay, constantDelay, fullJitterBackoff, limitRetries, recovering, retrying) import Data.ByteString.Base16 qualified as Base16 import Data.Text qualified as T import Hydra.Cardano.Api ( @@ -68,11 +68,11 @@ instance ChainBackend BlockfrostBackend where queryTip = withProject $ \_ prj -> Blockfrost.runBlockfrostM prj Blockfrost.queryTip - queryUTxO addresses = withProject $ \opts prj -> do + queryUTxO addresses = withProject $ \_ prj -> do Blockfrost.Genesis{_genesisNetworkMagic} <- Blockfrost.runBlockfrostM prj Blockfrost.queryGenesisParameters let networkId = Blockfrost.toCardanoNetworkId _genesisNetworkMagic - Blockfrost.runBlockfrostM prj $ Blockfrost.queryUTxO opts networkId addresses + Blockfrost.runBlockfrostM prj $ Blockfrost.queryUTxO networkId addresses queryUTxOByTxIn txins = withProject $ \opts prj -> do Blockfrost.Genesis{_genesisNetworkMagic} <- @@ -92,8 +92,8 @@ instance ChainBackend BlockfrostBackend where queryStakePools _ = withProject $ \_ prj -> Blockfrost.runBlockfrostM prj Blockfrost.queryStakePools - queryUTxOFor _ vk = withProject $ \opts prj -> - Blockfrost.runBlockfrostM prj $ Blockfrost.queryUTxOFor opts vk + queryUTxOFor _ vk = withProject $ \_ prj -> + Blockfrost.runBlockfrostM prj $ Blockfrost.queryUTxOFor vk submitTransaction tx = withProject $ \_ prj -> void $ Blockfrost.runBlockfrostM prj $ Blockfrost.submitTransaction tx @@ -106,10 +106,6 @@ instance ChainBackend BlockfrostBackend where Blockfrost.runBlockfrostM prj Blockfrost.queryGenesisParameters pure $ CardanoClient.computeBlockTime (fromInteger _genesisSlotLength) _genesisActiveSlotsCoefficient - getQueryDelay = BlockfrostBackend $ do - BlockfrostOptions{queryTimeout} <- ask - pure $ fromIntegral queryTimeout - withProject :: (BlockfrostOptions -> Blockfrost.Project -> IO a) -> BlockfrostBackend a withProject f = BlockfrostBackend $ do opts@BlockfrostOptions{projectPath} <- ask @@ -162,7 +158,7 @@ withBlockfrostChain opts tracer config ctx wallet chainStateHistory callback act ( "blockfrost-chain-connection" , handle onIOException $ do prj <- Blockfrost.projectFromFile projectPath - blockfrostChain tracer queue prj prefix handler wallet (runBlockfrostBackend opts getBlockTime) + blockfrostChain tracer queue prj prefix handler wallet ) ("blockfrost-chain-handle", action chainHandle) case res of @@ -203,13 +199,12 @@ blockfrostChain :: NonEmpty ChainPoint -> ChainSyncHandler m -> TinyWallet m -> - m NominalDiffTime -> m () -blockfrostChain tracer queue prj prefix handler wallet queryBlockTime = do +blockfrostChain tracer queue prj prefix handler wallet = do forever $ raceLabelled_ ("blockfrost-chain-follow", blockfrostChainFollow tracer prj prefix handler wallet) - ("blockfrost-submission", blockfrostSubmissionClient prj tracer queryBlockTime queue) + ("blockfrost-submission", blockfrostSubmissionClient tracer (submitViaBlockfrost prj) queue) blockfrostChainFollow :: forall m. @@ -221,10 +216,10 @@ blockfrostChainFollow :: TinyWallet m -> m () blockfrostChainFollow tracer prj prefix handler wallet = do - -- Genesis query and initial catch-up are both wrapped in retry to survive + -- Genesis query and start point resolution are wrapped in retry to survive -- transient HTTP errors (e.g. 403 rate limiting, connection resets). (blockTime, stateTVar) <- - retryOnBlockfrostError tracer maxRetries $ \_ -> do + retryOnBlockfrostError tracer blockfrostRetryPolicy $ \_ -> do Blockfrost.Genesis{_genesisSlotLength, _genesisActiveSlotsCoefficient} <- Blockfrost.runBlockfrostM prj Blockfrost.getLedgerGenesis let blockTime :: Double = realToFrac _genesisSlotLength / realToFrac _genesisActiveSlotsCoefficient @@ -232,7 +227,6 @@ blockfrostChainFollow tracer prj prefix handler wallet = do -- If none of them can be resolved, we fall back to the tip of the chain. blockHash <- resolvePrefixPoints (toList prefix) stateTVar <- newLabelledTVarIO "blockfrost-chain-state" blockHash - void $ catchUpToLatest blockHash stateTVar pure (blockTime, stateTVar) void $ @@ -249,39 +243,21 @@ blockfrostChainFollow tracer prj prefix handler wallet = do retryPolicy :: Double -> RetryPolicyM m retryPolicy blockTime' = constantDelay (truncate blockTime' * 1000 * 1000) - catchUpToLatest currentHash stateTVar = do - latestBlock <- Blockfrost.runBlockfrostM prj BlockfrostAPI.getLatestBlock - let targetHash = BlockfrostAPI._blockHash latestBlock - - catchUpLoop currentHash targetHash stateTVar - - catchUpLoop currentHash targetHash stateTVar = do - if currentHash == targetHash - then do - pure currentHash - else do - nextBlockHash <- rollForward tracer prj handler wallet 0 currentHash - atomically $ writeTVar stateTVar nextBlockHash - - if nextBlockHash == targetHash - then do - pure nextBlockHash - else catchUpLoop nextBlockHash targetHash stateTVar - + -- Process every already-confirmed successor of the last processed block, + -- then sleep one block time only once we caught up to the tip. Blocks with + -- zero confirmations are left for a later iteration: we only ever observe + -- blocks that have at least one successor. pollForNewBlocks blockTime' stateTVar = do - threadDelay (realToFrac blockTime') current <- readTVarIO stateTVar - nextBlockHash <- - rollForward tracer prj handler wallet 1 current - `catch` \case - MissingNextBlockHash{} -> do - pure current - ex -> throwIO ex - - when (nextBlockHash /= current) $ - atomically $ - writeTVar stateTVar nextBlockHash - + blocks <- + Blockfrost.runBlockfrostM prj $ + BlockfrostAPI.getNextBlocks' (Right current) (BlockfrostAPI.paged maxBlockBatch 1) + let confirmed = filter ((>= 1) . Blockfrost._blockConfirmations) blocks + forM_ confirmed $ \block -> do + processBlock tracer prj handler wallet block + atomically $ writeTVar stateTVar (Blockfrost._blockHash block) + when (length blocks < maxBlockBatch) $ + threadDelay (realToFrac blockTime') pollForNewBlocks blockTime' stateTVar resolvePrefixPoints :: [ChainPoint] -> m Blockfrost.BlockHash @@ -312,83 +288,70 @@ blockfrostChainFollow tracer prj prefix handler wallet = do ChainPoint _ headerHash -> pure $ Blockfrost.BlockHash (decodeUtf8 . Base16.encode . serialiseToRawBytes $ headerHash) -rollForward :: +processBlock :: (MonadIO m, MonadThrow m) => Tracer m CardanoChainLog -> Blockfrost.Project -> ChainSyncHandler m -> TinyWallet m -> - Integer -> - Blockfrost.BlockHash -> - m Blockfrost.BlockHash -rollForward tracer prj handler wallet blockConfirmations blockHash = do - block@Blockfrost.Block - { _blockHash - , _blockConfirmations - , _blockNextBlock - , _blockHeight - , _blockSlot - , _blockTime - } <- - Blockfrost.runBlockfrostM prj $ Blockfrost.getBlock (Right blockHash) - - -- Check if block within the safe zone to be processes - when (_blockConfirmations < blockConfirmations) $ - throwIO (NotEnoughBlockConfirmations _blockHash) - - -- Search block transactions - txHashesCBOR <- Blockfrost.runBlockfrostM prj . Blockfrost.allPages $ \p -> - Blockfrost.getBlockTxsCBOR' (Right _blockHash) p Blockfrost.def - - -- Check if block contains a reference to its next - nextBlockHash <- maybe (throwIO $ MissingNextBlockHash _blockHash) pure _blockNextBlock - - -- Convert to cardano-api Tx - receivedTxs <- mapM (toTx . (\(Blockfrost.TxHashCBOR (_txHash, cbor)) -> cbor)) txHashesCBOR + Blockfrost.Block -> + m () +processBlock tracer prj handler wallet block@Blockfrost.Block{_blockHash, _blockTxCount, _blockHeight, _blockSlot} = do + -- A block's transactions are a separate paginated request; the header + -- already tells us when there is nothing to fetch. + receivedTxs <- + if _blockTxCount == 0 + then pure [] + else do + txHashesCBOR <- + Blockfrost.runBlockfrostM prj . Blockfrost.allPages $ \p -> + Blockfrost.getBlockTxsCBOR' (Right _blockHash) p Blockfrost.def + mapM (toTx . (\(Blockfrost.TxHashCBOR (_txHash, cbor)) -> cbor)) txHashesCBOR let receivedTxIds = getTxId . getTxBody <$> receivedTxs let point = toChainPoint block traceWith tracer RolledForward{point, receivedTxIds} blockNo <- maybe (throwIO $ MissingBlockNo _blockHash) (pure . fromInteger) _blockHeight - let Blockfrost.BlockHash blockHash' = _blockHash - let blockHash'' = fromString $ T.unpack blockHash' blockSlot <- maybe (throwIO $ MissingBlockSlot _blockSlot) (pure . fromInteger . Blockfrost.unSlot) _blockSlot - let header = BlockHeader (SlotNo blockSlot) blockHash'' blockNo - -- wallet update + let Blockfrost.BlockHash blockHashText = _blockHash + let header = BlockHeader (SlotNo blockSlot) (fromString $ T.unpack blockHashText) blockNo update wallet header receivedTxs - onRollForward handler header receivedTxs - pure nextBlockHash - blockfrostSubmissionClient :: forall m. - (MonadIO m, MonadDelay m, MonadSTM m) => - Blockfrost.Project -> + MonadSTM m => Tracer m CardanoChainLog -> - -- | Action returning the chain's average block time (seconds), used to size - -- the delay before reporting 'PostTxError'. - m NominalDiffTime -> + -- | How to submit a transaction, yielding a rendered failure reason or the + -- transaction hash. Must not throw. + (Tx -> m (Either Text Blockfrost.TxHash)) -> TQueue m (Tx, TMVar m (Maybe (PostTxError Tx))) -> m () -blockfrostSubmissionClient prj tracer queryBlockTime queue = bfClient +blockfrostSubmissionClient tracer submit queue = bfClient where bfClient = do (tx, response) <- atomically $ readTQueue queue let txId = getTxId $ getTxBody tx traceWith tracer PostingTx{txId} - res <- liftIO $ Blockfrost.tryError $ Blockfrost.runBlockfrost prj $ Blockfrost.submitTransaction tx + res <- submit tx case res of Left err -> do - let postTxError = FailedToPostTx{failureReason = show err, failingTx = tx} + let postTxError = FailedToPostTx{failureReason = err, failingTx = tx} traceWith tracer PostingFailed{tx, postTxError} - blockTime <- queryBlockTime - threadDelay (realToFrac blockTime) atomically (putTMVar response (Just postTxError)) Right _ -> do traceWith tracer PostedTx{txId} atomically (putTMVar response Nothing) - bfClient + bfClient + +-- | Submit a transaction via Blockfrost, rendering both transport and API +-- level failures into a reason. +submitViaBlockfrost :: MonadIO m => Blockfrost.Project -> Tx -> m (Either Text Blockfrost.TxHash) +submitViaBlockfrost prj tx = + liftIO $ + (Right <$> Blockfrost.runBlockfrostM prj (Blockfrost.submitTransaction tx)) + `catch` (\(e :: APIBlockfrostError) -> pure . Left $ show e) + `catch` (\(e :: IOException) -> pure . Left $ show e) toChainPoint :: Blockfrost.Block -> ChainPoint toChainPoint Blockfrost.Block{_blockSlot, _blockHash} = @@ -406,18 +369,25 @@ toChainPoint Blockfrost.Block{_blockSlot, _blockHash} = maxRetries :: Int maxRetries = 10 --- | Retry an action on transient 'APIBlockfrostError' exceptions with --- exponential backoff (1s, 2s, 4s, ... capped at 60s). Gives up after --- the specified number of retries and re-throws the last exception. +-- | Maximum number of blocks fetched per poll iteration (the Blockfrost page +-- size limit). +maxBlockBatch :: Int +maxBlockBatch = 100 + +-- | Retry policy for transient Blockfrost errors: full-jitter exponential +-- backoff with 1s base, capped at 60s, at most 'maxRetries' retries. +blockfrostRetryPolicy :: MonadIO m => RetryPolicyM m +blockfrostRetryPolicy = capDelay 60_000_000 (fullJitterBackoff 1_000_000) <> limitRetries maxRetries + retryOnBlockfrostError :: (MonadIO m, Catch.MonadMask m) => Tracer m CardanoChainLog -> - Int -> + RetryPolicyM m -> (RetryStatus -> m a) -> m a -retryOnBlockfrostError tracer maxRetryCount = +retryOnBlockfrostError tracer policy = recovering - (fullJitterBackoff 2_000 <> limitRetries maxRetryCount) + policy [ \RetryStatus{rsCumulativeDelay} -> Handler $ \(ex :: APIBlockfrostError) -> do traceWith tracer $ BlockfrostTransientError{reason = show ex, retryDelay = rsCumulativeDelay} pure (isRetryable ex) diff --git a/hydra-node/src/Hydra/Chain/Blockfrost/Client.hs b/hydra-node/src/Hydra/Chain/Blockfrost/Client.hs index dfdda525da1..bc712a12116 100644 --- a/hydra-node/src/Hydra/Chain/Blockfrost/Client.hs +++ b/hydra-node/src/Hydra/Chain/Blockfrost/Client.hs @@ -15,6 +15,7 @@ import Blockfrost.Client ( Project, Slot (..), TransactionCBOR (..), + TxHash (..), TxHashCBOR (..), allPages, def, @@ -33,7 +34,7 @@ import Cardano.Chain.Genesis (mainnetProtocolMagicId) import Cardano.Crypto.ProtocolMagic (ProtocolMagicId (..)) import Data.Map.Strict qualified as Map import Data.Time.Clock.POSIX -import Hydra.Cardano.Api hiding (LedgerState, fromNetworkMagic, queryGenesisParameters) +import Hydra.Cardano.Api hiding (LedgerState, fromNetworkMagic, queryGenesisParameters, txId) import Cardano.Api.UTxO qualified as UTxO import Cardano.Ledger.Api.PParams @@ -64,7 +65,7 @@ import Data.Set qualified as Set import Data.Text qualified as T import Hydra.Cardano.Api.Prelude (fromNetworkMagic) import Hydra.Options (BlockfrostOptions (..)) -import Hydra.Tx (ScriptRegistry, newScriptRegistry) +import Hydra.Tx (ScriptRegistry, newScriptRegistry, txId) import Money qualified import Ouroboros.Consensus.Block (GenesisWindow (..)) import Ouroboros.Consensus.HardFork.History (Bound (..), EraEnd (..), EraParams (..), EraSummary (..), SafeZone (..), Summary (..), mkInterpreter, pattern NoPerasEnabled) @@ -84,10 +85,9 @@ data APIBlockfrostError = BlockfrostError Text | BlockfrostClientError BlockfrostException | DecodeError Text - | NotEnoughBlockConfirmations BlockHash | MissingBlockNo BlockHash | MissingBlockSlot (Maybe Slot) - | MissingNextBlockHash BlockHash + | BlockfrostRateLimited deriving stock (Show) deriving anyclass (Exception) @@ -96,21 +96,39 @@ isRetryable = \case BlockfrostError _ -> True BlockfrostClientError _ -> False DecodeError _ -> True - NotEnoughBlockConfirmations _ -> True MissingBlockNo _ -> True MissingBlockSlot _ -> True - MissingNextBlockHash _ -> True + BlockfrostRateLimited -> True +-- | 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) + +-- | Delay before the n-th rate-limit retry. +rateLimitBackoff :: Int -> DiffTime +rateLimitBackoff attempt = min 60 (2 ^ attempt) + +-- | How often to retry a rate-limited request before giving up. +maxRateLimitRetries :: Int +maxRateLimitRetries = 6 -- | Query for 'TxIn's in the search for outputs containing all the reference -- scripts of the 'ScriptRegistry'. @@ -476,14 +494,11 @@ queryScript scriptHashTxt = do -- | Query the Blockfrost API for address UTxO and convert to cardano 'UTxO'. -- NOTE: We accept the address list here to be compatible with cardano-api but in -- fact this is a single address query always. -queryUTxO :: BlockfrostOptions -> NetworkId -> [Address ShelleyAddr] -> BlockfrostClientT IO UTxO -queryUTxO BlockfrostOptions{queryTimeout} networkId addresses = do - -- NOTE: We can't know at the time of doing a query if the information on specific address UTxO is _fresh_ or not - -- so we try to wait for sufficient period of time and hope for best. - liftIO $ threadDelay $ fromIntegral queryTimeout +queryUTxO :: NetworkId -> [Address ShelleyAddr] -> BlockfrostClientT IO UTxO +queryUTxO networkId addresses = do let address = Blockfrost.Address . serialiseAddress $ List.head addresses utxoWithAddresses <- - Blockfrost.getAddressUtxos address + Blockfrost.allPages (\p -> Blockfrost.getAddressUtxos' address p Blockfrost.def) `catchError` \case Blockfrost.BlockfrostNotFound _ -> pure [] @@ -506,8 +521,8 @@ queryUTxO BlockfrostOptions{queryTimeout} networkId addresses = do ) utxoWithAddresses -queryUTxOFor :: BlockfrostOptions -> VerificationKey PaymentKey -> BlockfrostClientT IO UTxO -queryUTxOFor cfg vk = do +queryUTxOFor :: VerificationKey PaymentKey -> BlockfrostClientT IO UTxO +queryUTxOFor vk = do Blockfrost.Genesis { _genesisNetworkMagic = networkMagic } <- @@ -515,7 +530,7 @@ queryUTxOFor cfg vk = do let networkId = toCardanoNetworkId networkMagic case mkVkAddress networkId vk of ShelleyAddressInEra addr -> - queryUTxO cfg networkId [addr] + queryUTxO networkId [addr] ByronAddressInEra{} -> liftIO $ throwIO $ BlockfrostClientError ByronAddressNotSupported @@ -562,28 +577,50 @@ awaitTransaction :: BlockfrostOptions -> Tx -> VerificationKey PaymentKey -> Blo awaitTransaction cfg tx vk = do Blockfrost.Genesis{_genesisNetworkMagic} <- queryGenesisParameters let networkId = toCardanoNetworkId _genesisNetworkMagic - awaitUTxO networkId [makeShelleyAddress networkId (PaymentCredentialByKey $ verificationKeyHash vk) NoStakeAddress] (getTxId $ getTxBody tx) cfg + awaitUTxO networkId [makeShelleyAddress networkId (PaymentCredentialByKey $ verificationKeyHash vk) NoStakeAddress] tx cfg --- | Await for specific UTxO at address - the one that is produced by the given 'TxId'. +-- | Await inclusion of the given transaction and then wait until the +-- address query reflects its outputs at the given addresses. Return +-- those outputs (empty if the transaction pays nothing to them, in +-- which case only inclusion is awaited). awaitUTxO :: -- | Network id NetworkId -> - -- | Address we are interested in + -- | Addresses we are interested in [Address ShelleyAddr] -> - -- | Last transaction ID to await - TxId -> + -- | Transaction to await + Tx -> BlockfrostOptions -> BlockfrostClientT IO UTxO -awaitUTxO networkId addresses txid cfg@BlockfrostOptions{retryTimeout} = do - go retryTimeout +awaitUTxO networkId addresses tx BlockfrostOptions{retryTimeout} = do + awaitIncluded retryTimeout + unless (UTxO.null wantedUTxO) $ awaitVisible retryTimeout + pure wantedUTxO where - go 0 = liftIO $ throwIO $ BlockfrostClientError (TimeoutOnUTxO txid) - go n = do - utxo <- Blockfrost.tryError $ queryUTxO cfg networkId addresses - case utxo of - Left _e -> liftIO (threadDelay 1) >> go (n - 1) - Right utxo' -> - let wantedUTxO = UTxO.fromList $ List.filter (\(TxIn txid' _, _) -> txid' == txid) (UTxO.toList utxo') - in if UTxO.null wantedUTxO - then liftIO (threadDelay 1) >> go (n - 1) - else pure utxo' + txid = txId tx + + wantedUTxO = + UTxO.filter + ( \(TxOut addr _ _ _) -> case addr of + ShelleyAddressInEra sh -> sh `elem` addresses + ByronAddressInEra{} -> False + ) + (utxoFromTx tx) + + awaitIncluded 0 = liftIO $ throwIO $ BlockfrostClientError (TimeoutOnUTxO txid) + awaitIncluded n = do + res <- Blockfrost.tryError $ Blockfrost.getTx (Blockfrost.TxHash $ serialiseToRawBytesHexText txid) + case res of + Left _e -> liftIO (threadDelay 1) >> awaitIncluded (n - 1) + Right _ -> pure () + + -- NOTE: The address endpoint lags the tx endpoint, so inclusion alone does + -- not guarantee the next address query reflects this tx. Wait until it does, + -- since callers build follow-up transactions from what they query next. + awaitVisible 0 = liftIO $ throwIO $ BlockfrostClientError (TimeoutOnUTxO txid) + awaitVisible n = do + res <- Blockfrost.tryError $ queryUTxO networkId addresses + case res of + Right utxo' + | UTxO.inputSet wantedUTxO `Set.isSubsetOf` UTxO.inputSet utxo' -> pure () + _ -> liftIO (threadDelay 1) >> awaitVisible (n - 1) diff --git a/hydra-node/src/Hydra/Chain/Direct.hs b/hydra-node/src/Hydra/Chain/Direct.hs index 49b41efde89..8178460fc98 100644 --- a/hydra-node/src/Hydra/Chain/Direct.hs +++ b/hydra-node/src/Hydra/Chain/Direct.hs @@ -133,8 +133,6 @@ instance ChainBackend DirectBackend where getBlockTime = withNodeConn $ \ci -> CardanoClient.queryBlockTime ci CardanoClient.QueryTip - getQueryDelay = pure 0 - withNodeConn :: (LocalNodeConnectInfo -> IO a) -> DirectBackend a withNodeConn f = DirectBackend $ do DirectOptions{networkId, nodeSocket} <- ask diff --git a/hydra-node/src/Hydra/Config.hs b/hydra-node/src/Hydra/Config.hs index e7bf8bc40a1..38f3a89ec6c 100644 --- a/hydra-node/src/Hydra/Config.hs +++ b/hydra-node/src/Hydra/Config.hs @@ -349,9 +349,8 @@ parseBlockfrostOptions :: Object -> Parser BlockfrostOptions parseBlockfrostOptions o = do checkUnknownKeys ["mode", "project-path", "query-timeout", "retry-timeout"] o projectPath <- o .:? "project-path" .!= defaultBlockfrostOptions.projectPath - queryTimeout <- o .:? "query-timeout" .!= defaultBlockfrostOptions.queryTimeout retryTimeout <- o .:? "retry-timeout" .!= defaultBlockfrostOptions.retryTimeout - pure BlockfrostOptions{projectPath, queryTimeout, retryTimeout} + pure BlockfrostOptions{projectPath, retryTimeout} -- --------------------------------------------------------------------------- -- Helpers @@ -542,7 +541,6 @@ renderConfig opts = object [ "mode" .= ("blockfrost" :: Text) , "project-path" .= o.projectPath - , "query-timeout" .= o.queryTimeout , "retry-timeout" .= o.retryTimeout ] diff --git a/hydra-node/src/Hydra/Options.hs b/hydra-node/src/Hydra/Options.hs index e9180d85793..9b2cd1a5369 100644 --- a/hydra-node/src/Hydra/Options.hs +++ b/hydra-node/src/Hydra/Options.hs @@ -186,7 +186,6 @@ data DirectOptions = DirectOptions data BlockfrostOptions = BlockfrostOptions { projectPath :: FilePath -- ^ Path to the blockfrost project file - , queryTimeout :: Int , retryTimeout :: Int } deriving stock (Generic, Show, Eq) @@ -196,13 +195,9 @@ defaultBlockfrostOptions :: BlockfrostOptions defaultBlockfrostOptions = BlockfrostOptions { projectPath = "blockfrost-project.txt" - , queryTimeout = defaultBFQueryTimeout , retryTimeout = defaultBFRetryTimeout } -defaultBFQueryTimeout :: Int -defaultBFQueryTimeout = 30 - defaultBFRetryTimeout :: Int defaultBFRetryTimeout = 300 @@ -377,7 +372,6 @@ instance Semigroup BlockfrostOptions where base <> cli = BlockfrostOptions { projectPath = o defaultBlockfrostOptions.projectPath base.projectPath cli.projectPath - , queryTimeout = o defaultBlockfrostOptions.queryTimeout base.queryTimeout cli.queryTimeout , retryTimeout = o defaultBlockfrostOptions.retryTimeout base.retryTimeout cli.retryTimeout } where @@ -468,7 +462,6 @@ chainBackendOptionsParser = directOptionsParser <|> blockfrostOptionsParser fmap Blockfrost $ BlockfrostOptions <$> blockfrostProjectPathParser - <*> blockfrostQueryTimeoutParser <*> blockfrostRetryTimeoutParser newtype GenerateKeyPair = GenerateKeyPair @@ -664,17 +657,6 @@ blockfrostProjectPathParser = "Blockfrost project path containing the api key." ) -blockfrostQueryTimeoutParser :: Parser Int -blockfrostQueryTimeoutParser = - option - auto - ( long "blockfrost-query-timeout" - <> metavar "SECONDS" - <> value defaultBFQueryTimeout - <> showDefault - <> help "Timeout for single queries to the Blockfrost API, in seconds." - ) - blockfrostRetryTimeoutParser :: Parser Int blockfrostRetryTimeoutParser = option @@ -1236,9 +1218,8 @@ toArgs , chainBackendOptions } -> ( case chainBackendOptions of - Blockfrost BlockfrostOptions{projectPath, queryTimeout, retryTimeout} -> + Blockfrost BlockfrostOptions{projectPath, retryTimeout} -> ["--blockfrost", projectPath] - <> ["--blockfrost-query-timeout", show queryTimeout] <> ["--blockfrost-retry-timeout", show retryTimeout] Direct DirectOptions{networkId, nodeSocket} -> toArgNetworkId networkId diff --git a/hydra-node/test/Hydra/Chain/BlockfrostSpec.hs b/hydra-node/test/Hydra/Chain/BlockfrostSpec.hs index 7078db4e805..38ca1910ecc 100644 --- a/hydra-node/test/Hydra/Chain/BlockfrostSpec.hs +++ b/hydra-node/test/Hydra/Chain/BlockfrostSpec.hs @@ -3,11 +3,19 @@ module Hydra.Chain.BlockfrostSpec where import Hydra.Prelude import Test.Hspec +import Control.Concurrent.Class.MonadSTM (takeTMVar, writeTQueue) +import Control.Retry (RetryPolicyM, limitRetries) import Control.Tracer (nullTracer) -import Hydra.Chain.Blockfrost (retryOnBlockfrostError) -import Hydra.Chain.Blockfrost.Client (APIBlockfrostError (..), BlockfrostException (..), isRetryable) +import Hydra.Chain.Blockfrost (blockfrostSubmissionClient, retryOnBlockfrostError) +import Hydra.Chain.Blockfrost.Client (APIBlockfrostError (..), BlockfrostException (..), TxHash (..), isRetryable, rateLimitBackoff) import Hydra.Chain.Direct.Handlers (CardanoChainLog) import Hydra.Logging (Tracer) +import Test.Hydra.Prelude (failAfter) +import Test.Hydra.Tx.Gen () +import Test.QuickCheck (arbitrary, generate) + +retry :: RetryPolicyM IO +retry = limitRetries 3 spec :: Spec spec = do @@ -18,11 +26,14 @@ spec = do it "treats BlockfrostError as retryable" $ do isRetryable (BlockfrostError "some API error") `shouldBe` True + it "treats BlockfrostRateLimited as retryable" $ do + isRetryable BlockfrostRateLimited `shouldBe` True + describe "retryOnBlockfrostError" $ do it "retries on transient APIBlockfrostError and eventually succeeds" $ do attemptsRef <- newIORef (0 :: Int) result <- - retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) 3 $ const $ do + retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) retry $ const $ do attempts <- readIORef attemptsRef writeIORef attemptsRef (attempts + 1) if attempts < 2 @@ -37,7 +48,7 @@ spec = do let action = do modifyIORef attemptsRef (+ 1) throwIO $ BlockfrostError "persistent error" - retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) 3 (const action) + retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) retry (const action) `shouldThrow` \case BlockfrostError{} -> True _ -> False @@ -47,7 +58,7 @@ spec = do it "retries on HTTP error (BlockfrostError Text) and eventually succeeds" $ do attemptsRef <- newIORef (0 :: Int) result <- - retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) 3 $ const $ do + retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) retry $ const $ do attempts <- readIORef attemptsRef modifyIORef attemptsRef (+ 1) if attempts < 2 @@ -62,7 +73,7 @@ spec = do let action = do modifyIORef attemptsRef (+ 1) throwIO $ BlockfrostError "HTTP 403 Forbidden" - retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) 3 (const action) + retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) retry (const action) `shouldThrow` \case BlockfrostError _ -> True _ -> False @@ -74,9 +85,42 @@ spec = do let action = do modifyIORef attemptsRef (+ 1) throwIO $ BlockfrostClientError ByronAddressNotSupported - retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) 3 (const action) + retryOnBlockfrostError (nullTracer :: Tracer IO CardanoChainLog) retry (const action) `shouldThrow` \case BlockfrostClientError{} -> True _ -> False finalAttempts <- readIORef attemptsRef finalAttempts `shouldBe` 1 + + describe "rateLimitBackoff" $ + it "grows exponentially and caps at 60s" $ do + rateLimitBackoff 0 `shouldBe` 1 + rateLimitBackoff 3 `shouldBe` 8 + rateLimitBackoff 10 `shouldBe` 60 + + describe "blockfrostSubmissionClient" $ + it "reports submission failures immediately and keeps serving the queue" $ + failAfter 5 $ do + queue <- newLabelledTQueueIO "test-submission-queue" + calls <- newIORef (0 :: Int) + let submit _tx = do + n <- atomicModifyIORef' calls $ \c -> (c + 1, c) + pure $ + if n == 0 + then Left "submission failed" + else Right (TxHash "deadbeef") + tx1 <- generate arbitrary + tx2 <- generate arbitrary + withAsyncLabelled ("blockfrost-submit", blockfrostSubmissionClient (nullTracer :: Tracer IO CardanoChainLog) submit queue) $ \_ -> do + res1 <- postViaQueue queue tx1 + res1 `shouldSatisfy` isJust + res2 <- postViaQueue queue tx2 + res2 `shouldSatisfy` isNothing + where + postViaQueue :: forall m a b. MonadLabelledSTM m => TQueue m (a, TMVar m b) -> a -> m b + postViaQueue queue tx = do + response <- atomically $ do + r <- newLabelledEmptyTMVar "test-response" + writeTQueue queue (tx, r) + pure r + atomically $ takeTMVar response diff --git a/hydra-node/test/Hydra/Chain/ScriptRegistrySpec.hs b/hydra-node/test/Hydra/Chain/ScriptRegistrySpec.hs index 4c77005fa91..cfdf4aa29de 100644 --- a/hydra-node/test/Hydra/Chain/ScriptRegistrySpec.hs +++ b/hydra-node/test/Hydra/Chain/ScriptRegistrySpec.hs @@ -81,7 +81,6 @@ instance ChainBackend ATestBackend where submitTransaction _ = error "submitTransaction" awaitTransaction _ _ = error "awaitTransaction" getBlockTime = error "getBlockTime" - getQueryDelay = pure 0 queryNetworkId = pure Mainnet queryProtocolParameters _ = pure emptyPParams querySystemStart _ = SystemStart <$> liftIO getCurrentTime @@ -111,7 +110,6 @@ instance ChainBackend SuccessfulBackend where querySystemStart _ = SystemStart <$> liftIO getCurrentTime queryEraHistory _ = pure eraHistoryWithoutHorizon queryStakePools _ = pure mempty - getQueryDelay = pure 0 -- Other methods are not needed for this test. queryGenesisParameters = error "queryGenesisParameters" diff --git a/hydra-node/test/Hydra/OptionsSpec.hs b/hydra-node/test/Hydra/OptionsSpec.hs index 521e3f791b1..59da051cf7f 100644 --- a/hydra-node/test/Hydra/OptionsSpec.hs +++ b/hydra-node/test/Hydra/OptionsSpec.hs @@ -388,8 +388,6 @@ spec = parallel $ it "parses --blockfrost with timeouts" $ [ "--blockfrost" , "blockfrost-project.txt" - , "--blockfrost-query-timeout" - , "30" , "--blockfrost-retry-timeout" , "600" ] @@ -401,8 +399,7 @@ spec = parallel $ & #chainBackendOptions .~ Blockfrost defaultBlockfrostOptions - { queryTimeout = 30 - , retryTimeout = 600 + { retryTimeout = 600 } ) } @@ -519,7 +516,6 @@ spec = parallel $ Blockfrost BlockfrostOptions { projectPath = "baz" - , queryTimeout = 30 , retryTimeout = 300 } , publishSigningKey = "cardano.sk" diff --git a/hydra-tui/src/Hydra/TUI.hs b/hydra-tui/src/Hydra/TUI.hs index 9f2521bbac4..16639d5afc0 100644 --- a/hydra-tui/src/Hydra/TUI.hs +++ b/hydra-tui/src/Hydra/TUI.hs @@ -25,7 +25,6 @@ import Hydra.Chain.CardanoClient as CC import Hydra.Chain.Direct.State () import Hydra.Client (HydraEvent (..), withClient) import Hydra.Node.Util (readFileTextEnvelopeThrow) -import Hydra.Options (BlockfrostOptions (..), defaultBFQueryTimeout, defaultBFRetryTimeout) import Hydra.TUI.Config (Theme (..), TuiConfig (..), readConfig) import Hydra.TUI.Drawing import Hydra.TUI.Handlers @@ -49,14 +48,7 @@ mkBFClient networkId bfProject = CardanoClient { queryUTxOByAddress = \address -> do prj <- liftIO $ BF.projectFromFile bfProject - let bfOptions = - BlockfrostOptions - { projectPath = bfProject - , queryTimeout = defaultBFQueryTimeout - , retryTimeout = defaultBFRetryTimeout - } - - BF.runBlockfrostM prj $ BF.queryUTxO bfOptions networkId address + BF.runBlockfrostM prj $ BF.queryUTxO networkId address , networkId }