haskoin-node 2.3.1 → 2.4.0

raw patch · 35 files changed

+5486/−1931 lines, 35 filesdep +QuickCheckdep +directorydep +processdep −hashabledep −mtldep −safedep ~asyncdep ~basedep ~bytestringsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: QuickCheck, directory, process

Dependencies removed: hashable, mtl, safe, unordered-containers

Dependency ranges changed: async, base, bytestring, cereal, conduit, conduit-extra, containers, fast-logger, haskoin-core, hspec, network, nqe, random, rocksdb-haskell-jprupp, rocksdb-query, stm, string-conversions, text, time, transformers

API changes (from Hackage documentation)

Files

• CHANGELOG.md +93/−0
• README.md +121/−20
• Setup.hs +0/−2
• haskoin-node.cabal +104/−44
• src/Haskoin/Node.hs +68/−68
• src/Haskoin/Node/Address.hs +157/−0
• src/Haskoin/Node/Chain.hs +309/−525
• src/Haskoin/Node/Chain/Database.hs +169/−0
• src/Haskoin/Node/Chain/Sync.hs +204/−0
• src/Haskoin/Node/Logger.hs +5/−11
• src/Haskoin/Node/Peer.hs +145/−147
• src/Haskoin/Node/Peer/Lock.hs +102/−0
• src/Haskoin/Node/Peer/Request.hs +104/−0
• src/Haskoin/Node/PeerMgr.hs +332/−561
• src/Haskoin/Node/PeerMgr/Addresses.hs +208/−0
• src/Haskoin/Node/PeerMgr/Handshake.hs +142/−0
• src/Haskoin/Node/PeerMgr/Lifecycle.hs +209/−0
• test/Haskoin/NodeSpec.hs +0/−552
• test/Spec.hs +0/−1
• test/common/Haskoin/Node/Test.hs +391/−0
• test/regtest/Containerfile +49/−0
• test/regtest/Main.hs +196/−0
• test/regtest/Regtest/Podman.hs +231/−0
• test/unit/Haskoin/Node/AddressSpec.hs +45/−0
• test/unit/Haskoin/Node/Chain/SyncSpec.hs +154/−0
• test/unit/Haskoin/Node/ChainSpec.hs +407/−0
• test/unit/Haskoin/Node/Peer/LockSpec.hs +60/−0
• test/unit/Haskoin/Node/Peer/RequestSpec.hs +90/−0
• test/unit/Haskoin/Node/PeerMgr/AddressesSpec.hs +139/−0
• test/unit/Haskoin/Node/PeerMgr/HandshakeSpec.hs +95/−0
• test/unit/Haskoin/Node/PeerMgr/LifecycleSpec.hs +157/−0
• test/unit/Haskoin/Node/PeerMgrSpec.hs +228/−0
• test/unit/Haskoin/Node/PeerSpec.hs +246/−0
• test/unit/Haskoin/NodeSpec.hs +525/−0
• test/unit/Spec.hs +1/−0

@@ -4,6 +4,99 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [2.4.0] - 2026-11-09++### Added++- `chainBlocksAnnounced`: the node passes blocks announced in inventory+  messages to the chain, which asks the peer for headers if it does not know+  them.+- Pure state machines, each with property tests: the life of a connection+  (`Haskoin.Node.PeerMgr.Lifecycle`) and its version handshake+  (`Haskoin.Node.PeerMgr.Handshake`), the address book+  (`Haskoin.Node.PeerMgr.Addresses`), reserving a peer+  (`Haskoin.Node.Peer.Lock`), choosing which peer to ask for headers+  (`Haskoin.Node.Chain.Sync`), and receiving requested blocks and transactions+  (`Haskoin.Node.Peer.Request`).+- `isSynced` says whether the header chain counts as synced, as a pure+  function.+- `newOwner`, `reservePeer`, `releasePeer` and `peerHolder` reserve a peer for+  an owner, so that only that owner can release it.+- `PeerException` says why a peer connection ended, instead of an `error`+  message.+- `Haskoin.Node.Address` and `Haskoin.Node.Chain.Database` hold code split+  out of the peer manager, the node and the chain, which still re-export it.+  Every module is exposed, and `PeerMessage` is exported.+- Unit tests of the chain, peer and peer manager processes against in-memory+  mock peers.+- A regtest suite that runs the node against Bitcoin Core 32.1 and Bitcoin Cash+  Node 29.2.0 in Podman, from official binaries with checked hashes, and a CI+  job that runs it.+- `cabal.project`, so the package builds with Cabal too.+- `chainRunHeaders` runs header functions of haskoin-core, such as+  `getAncestor` or `blockLocator`, against the headers of a chain, and+  `runHeaders` does the same for a `ChainDB`.++### Changed++- If no headers are requested or received for `timeout` seconds, every+  connected peer is asked for headers again. This bounds how long any stall of+  the header chain can last.+- The chain checks for timeouts at intervals of at most a quarter of+  `timeout`, between 0.1 and 20 seconds, instead of every 2 to 20 seconds.+- Breaking: `OnlinePeer` holds the state of the connection in a `state` field+  instead of the `verack`, `online`, `version`, `ping`, `pings`, `connected`+  and `tickled` fields. These are still available as computed fields, so+  `o.version` and `o.online` keep working, but code that builds or pattern+  matches `OnlinePeer` records must change.+- Breaking: the busy flag of a `Peer` is a lock held by an owner, so `wrapPeer`+  takes a `TVar (Lock Owner)`. `setBusy` and `setFree` are deprecated: they act+  for a single owner shared by all their callers, and `setFree` no longer+  releases a peer that another owner reserved. The chain reserves peers as its+  own owner.+- A peer told to disconnect is no longer returned by `getPeers`, and its+  messages are no longer routed, even before its connection closes.+- A peer that does not complete the handshake within `timeout` seconds is+  disconnected, even if it keeps sending other messages.+- An address whose connection fails waits before it is tried again, twice as+  long after each failure up to ten minutes, and a relayed address whose peer+  broke the protocol waits a day. Before, failing addresses were retried+  within seconds forever. The configured peers and seeds are resolved again at+  most every five minutes.+- Breaking: `ChainT` and `runChainT` are removed, along with the+  `BlockHeaders` instances for `ReaderT Chain m` and `ReaderT ChainConfig m`.+  Use `chainRunHeaders` instead. The package no longer uses monad+  transformers outside `Haskoin.Node.Chain.Database`, where a `ReaderT` carries+  the database for haskoin-core's header functions; `withNode` nests its+  resources directly instead of through `ContT`.+- Breaking: every function runs in `IO`. The `MonadIO` and `MonadUnliftIO`+  constraints are gone, so code in another monad must call the node with+  `liftIO`. The library depends on async and stm instead of unliftio.+- The package uses the GHC2024 language edition, strict warnings and upper+  bounds on its dependencies. `Setup.hs`, `hie.yaml` and `.hindent.yaml` were+  removed.+- `stack.yaml` builds against secp256k1-haskell 2.0.1. The README's Stack+  example names the same version.++### Fixed++- Peers were ranked by the newest ping and the ten fastest ever measured,+  instead of the last eleven: slow recent pings were dropped. A peer that+  became slow kept ranking as fast.+- The header chain stopped following the network when a peer announced new+  blocks with an inventory message instead of headers, which peers do when+  many blocks are found at once. The chain stayed at the old height until the+  peer reconnected.+- A peer that announced headers that did not connect to the chain was+  disconnected. It is now asked for headers.+- A syncing peer that timed out kept the chain waiting until its disconnection+  arrived. The chain now moves on at once.+- A full batch of headers from a peer that was not being synced against was+  not followed up.+- `getData`, `getBlocks` and `getTxs` failed when the peer sent an unrelated+  message, or a notfound message for items that were not requested, before all+  the requested items arrived.+ ## [2.3.1] - 2026-11-07  ### Changed

@@ -70,6 +70,18 @@ - **Small, isolated components.** The peer manager, the header chain and each   connection are separate [NQE](https://github.com/jprupp/nqe) processes with   their own state. One misbehaving peer dies alone.+- **Explicit protocol states.** What the node expects from each peer, during the+  handshake, while syncing headers and while receiving requested data, is+  written down as a pure state machine. A message that does not fit the current+  state is refused instead of handled by accident, and every rule is tested+  without a network.+- **Plain `IO`.** Every function runs in `IO`, with no monad transformers and+  no `MonadIO` or `MonadUnliftIO` constraints. Types say exactly what runs, GHC+  has nothing to specialise, and long-running loops cannot leak through an+  unspecialised dictionary. From your own monad, call the node with `liftIO`.+- **Never stuck.** Every hint that a peer has blocks the node lacks makes it ask+  for headers, and a quiet period makes it ask every peer, so a missed+  announcement cannot leave the chain behind.  ## Features @@ -77,11 +89,11 @@ | --------------- | ---------------------------------------------------------------------------------------------- | | Networks        | Every network of haskoin-core: Bitcoin and Bitcoin Cash mainnet, testnets and regtest           | | Peer discovery  | DNS seeds and addresses relayed by peers, or a static list of peers                             |-| Peer management | Version handshake, ping-based latency, peers sorted by median ping, rotation after a maximum age |-| Header sync     | Headers downloaded from one peer at a time, connected to the chain with the most work, stored in RocksDB |+| Peer management | Version handshake, ping-based latency, peers sorted by median ping, rotation after a maximum age, retry backoff and bans for bad addresses |+| Header sync     | Headers downloaded from one peer at a time, connected to the chain with the most work, stored in RocksDB; new blocks followed whether peers announce them with headers or inventory, and peers polled after a quiet period | | Events          | Best header changes, sync completion, peer connections and every message received              | | Data requests   | Fetch blocks and transactions from a peer, with a timeout                                       |-| Protection      | Strict parsing, per-command size limits, back-pressure, bounded address pool                    |+| Protection      | Strict parsing, per-command size limits, back-pressure, bounded address pool, explicit handshake states |  ### What is not included @@ -150,9 +162,9 @@  ```yaml extra-deps:-  - haskoin-node-2.3.1+  - haskoin-node-2.4.0   - haskoin-core-1.6.1-  - secp256k1-haskell-2.0.0+  - secp256k1-haskell-2.0.1   - base16-2.0   - rocksdb-haskell-jprupp-2.6.1   - rocksdb-query-0.6.1@@ -312,11 +324,28 @@  - The **peer manager** (`Haskoin.Node.PeerMgr`) finds peer addresses, connects   to them, does the version handshake, pings them, and drops those that are slow,-  silent, misbehaving or too old.+  silent, misbehaving or too old. An address whose connection failed is tried+  again later each time, and one whose peer misbehaved waits a day. - The **chain** (`Haskoin.Node.Chain`) asks one peer at a time for headers,   connects them to the headers it has, follows the chain with the most work, and-  stores the headers in RocksDB.+  stores the headers in RocksDB. It asks a peer for headers whenever the peer+  announces blocks it does not know, with headers or with inventory, and asks+  every peer when it has heard nothing for `timeout` seconds. +The rules for talking to peers are pure state machines, each in its own module:++| Module                           | Rules                                                          |+| -------------------------------- | -------------------------------------------------------------- |+| `Haskoin.Node.PeerMgr.Lifecycle` | Life of a connection: handshake, pings, timers, disconnection  |+| `Haskoin.Node.PeerMgr.Handshake` | The version handshake, the first stage of a connection         |+| `Haskoin.Node.PeerMgr.Addresses` | Which address to try next, and when to retry one               |+| `Haskoin.Node.Chain.Sync`        | Which peer to ask for headers, and when headers count as synced |+| `Haskoin.Node.Peer.Request`      | Receiving requested blocks and transactions                    |+| `Haskoin.Node.Peer.Lock`         | Who may hold a peer for a long conversation                    |++The processes store the state and carry out what a transition says. Time is an+input to the transitions, so every timing rule is tested without waiting.+ Each peer connection is its own process (`Haskoin.Node.Peer`), run by a supervisor inside the peer manager. @@ -353,7 +382,8 @@ | `0x92`     | data version                  |  You can keep your own data in the same database, as Haskoin Store does, as long-as your keys do not start with these bytes.+as your keys do not start with these bytes. `Haskoin.Node.Chain.Database` has+the code.  ## Usage guide @@ -371,9 +401,28 @@ ```  `getBlocks`, `getTxs` and `getData` return `Nothing` if the peer does not send-everything, in order, before the timeout in seconds. `pingPeer` measures a round-trip.+everything, in order, before the timeout in seconds. Unrelated messages that the+peer sends in between are ignored. `pingPeer` measures a round trip. +`getPeers` leaves out peers that are being disconnected, even before their+connection closes.++For a conversation that spans many messages, such as downloading a long run of+blocks, reserve the peer first so that no other task uses it at the same time.+The header chain does the same while it syncs. Only the owner that reserved a+peer can release it:++```haskell+owner <- newOwner "downloader"+reserved <- reservePeer owner p+when reserved $ do+  -- ... long conversation with p ...+  void (releasePeer owner p)+```++`setBusy` and `setFree` still work, but are deprecated: they act for a single+owner shared by all their callers.+ ### Querying headers  ```haskell@@ -385,7 +434,7 @@ synced <- chainIsSynced node.chain ``` -`runChainT` runs the header functions of haskoin-core, such as `getAncestor`,+`chainRunHeaders` runs the header functions of haskoin-core, such as `getAncestor`, against the database of the chain.  ### Static peers@@ -420,26 +469,43 @@ | `address`     | This node's address, sent in version messages                             | | `net`         | Network constants from haskoin-core (`btc`, `bch`, `btcTest`, ...)        | | `pub`         | Publisher that receives every `NodeEvent`                                 |-| `timeout`     | Seconds of silence, or without a pong, before a peer is dropped           |+| `timeout`     | Seconds of silence, without a pong, or without completing the handshake, before a peer is dropped; also how long the syncing peer may take to send headers, and the quiet period after which every peer is asked for headers | | `maxPeerLife` | Seconds after which a peer is dropped, so that connections rotate         | | `connect`     | How to open a connection; usually `withConnection`                        | | `logger`      | Where to log; see `withLogger`                                            |  ## Module overview -`Haskoin.Node` is the only exposed module. It re-exports:+`Haskoin.Node` re-exports everything a program needs:  | Part           | Contents                                                                    | | -------------- | --------------------------------------------------------------------------- | | Node           | `NodeConfig`, `NodeEvent`, `Node`, `withNode`, `withConnection`              |-| Peers          | `Peer`, `PeerEvent`, `sendMessage`, `getBlocks`, `getTxs`, `getData`, `pingPeer`, `killPeer` |-| Peer manager   | `PeerMgr`, `OnlinePeer`, `getPeers`, `getOnlinePeer`, `toSockAddr`, `isRoutable` |-| Chain          | `Chain`, `ChainEvent`, `chainGetBest`, `chainGetBlock`, `chainGetAncestor`, `chainGetParents`, `chainGetSplitBlock`, `chainBlockMain`, `chainIsSynced`, `runChainT` |+| Peers          | `Peer`, `PeerEvent`, `PeerException`, `sendMessage`, `getBlocks`, `getTxs`, `getData`, `pingPeer`, `killPeer`, `newOwner`, `reservePeer`, `releasePeer` |+| Peer manager   | `PeerMgr`, `OnlinePeer`, `PeerState`, `Stage`, `DisconnectReason`, `getPeers`, `getOnlinePeer`, `toSockAddr`, `isRoutable` |+| Chain          | `Chain`, `ChainEvent`, `chainGetBest`, `chainGetBlock`, `chainGetAncestor`, `chainGetParents`, `chainGetSplitBlock`, `chainBlockMain`, `chainIsSynced`, `chainRunHeaders` | | Logging        | `Logger`, `LogLevel`, `withLogger`, `withSource`, `logDebug` ... `logError`  |  The lower-level functions that feed messages to the peer manager and the chain are exported too, for programs that wire the components themselves. +Every module is exposed and can be imported on its own:++| Module                           | Purpose                                                     |+| -------------------------------- | ----------------------------------------------------------- |+| `Haskoin.Node.Peer`              | The process that talks to one peer, and data requests       |+| `Haskoin.Node.Peer.Request`      | State machine for receiving requested blocks and transactions |+| `Haskoin.Node.Peer.Lock`         | State machine for reserving a peer                          |+| `Haskoin.Node.PeerMgr`           | Finds, connects, ranks and drops peers                      |+| `Haskoin.Node.PeerMgr.Lifecycle` | State machine for the life of a connection                  |+| `Haskoin.Node.PeerMgr.Handshake` | State machine for the version handshake                     |+| `Haskoin.Node.PeerMgr.Addresses` | State machine for the address book                          |+| `Haskoin.Node.Chain`             | Keeps the header chain in sync                              |+| `Haskoin.Node.Chain.Sync`        | State machine that decides which peer to ask for headers, and when the chain is synced |+| `Haskoin.Node.Chain.Database`    | Storage of headers in RocksDB                               |+| `Haskoin.Node.Address`           | Parsing peer addresses and opening connections              |+| `Haskoin.Node.Logger`            | Logging                                                     |+ ## Security notes  - Every message header is checked before its payload is read. The payload limit@@ -447,11 +513,21 @@   for everything but block messages. - List messages are refused as soon as their count is over the limit, such as   2000 headers, 1000 addresses or 50000 inventory entries.-- A peer that sends anything invalid is disconnected.+- A peer that sends anything invalid is disconnected, with a `PeerException`+  saying why.+- A peer that sends a second version or verack, or does not complete the+  handshake within `timeout` seconds even while sending other messages, is+  disconnected.+- A peer that sends invalid headers, or does not answer a request for headers in+  time, is never asked for headers again before it disconnects. - Addresses relayed by peers are only used if they are public unicast   addresses, so a peer cannot make the node connect to services on its own host   or network. Static peers are not checked. - At most 4096 relayed addresses are kept.+- An address whose connection fails waits twice as long after each failure+  before it is tried again, up to ten minutes. A relayed address whose peer+  broke the protocol is not tried again for a day. Static peers are never+  banned, and wait at most ten minutes. - Headers are not reported as synced until they have the minimum chain work of   the network. Above that work, forks from far below the tip on a chain with   less work are refused.@@ -475,9 +551,34 @@ `package.yaml` is the source of the package description. Change it rather than `haskoin-node.cabal`, which [hpack](https://github.com/sol/hpack) generates. -The tests are in `test/` and use [hspec](https://hspec.github.io/). They run-against mock peers on the Bitcoin Cash regression test network, so they need no-Internet access.+The package also builds with [Cabal](https://www.haskell.org/cabal)+(`cabal build`, `cabal test`).++There are two test suites, using [hspec](https://hspec.github.io/) and+[QuickCheck](https://hackage.haskell.org/package/QuickCheck):++- `spec` (`test/unit`) tests the state machines with properties, and runs the+  peer, peer manager, chain and node against in-memory mock peers. It needs no+  Internet access.++  ```sh+  stack test haskoin-node:spec+  ```++- `regtest` (`test/regtest`) runs the node against+  [Bitcoin Core](https://bitcoincore.org) and+  [Bitcoin Cash Node](https://bitcoincashnode.org) in regression test mode. The+  nodes run in a [Podman](https://podman.io) container built from+  `test/regtest/Containerfile`, which installs the official release binaries on+  Ubuntu after checking their SHA-256 hashes. The image is built the first time+  the suite runs. Nodes use random ports, so runs do not collide. The suite is+  skipped if Podman is missing or `HASKOIN_NODE_SKIP_REGTEST` is set.++  ```sh+  stack test haskoin-node:regtest+  ```++Set `HASKOIN_TEST_LOG` to see the node's logs while tests run.  ### Contributing 

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

@@ -1,105 +1,165 @@-cabal-version: 1.12+cabal-version: 1.18  -- This file has been generated from package.yaml by hpack version 0.39.6. -- -- see: https://github.com/sol/hpack  name:           haskoin-node-version:        2.3.1+version:        2.4.0 synopsis:       P2P library for Bitcoin and Bitcoin Cash-description:    Please see the README on GitHub at <https://github.com/jprupp/haskoin-node#readme>+description:    Connect to Bitcoin (BTC) and Bitcoin Cash (BCH) peers, keep them healthy, and sync block headers into a RocksDB database. Blocks and transactions can be downloaded from connected peers. See the README on GitHub at <https://github.com/jprupp/haskoin-node#readme>. category:       Bitcoin, Finance, Network-homepage:       http://github.com/jprupp/haskoin-node#readme-bug-reports:    http://github.com/jprupp/haskoin-node/issues+homepage:       https://github.com/jprupp/haskoin-node#readme+bug-reports:    https://github.com/jprupp/haskoin-node/issues author:         JP Rupp maintainer:     jprupp@protonmail.ch+copyright:      2020 Haskoin Developers license:        MIT license-file:   LICENSE build-type:     Simple+tested-with:+    GHC == 9.12.4 extra-source-files:+    test/regtest/Containerfile+extra-doc-files:     README.md     CHANGELOG.md  source-repository head   type: git-  location: https://github.com/jprupp/haskoin-node.git+  location: https://github.com/jprupp/haskoin-node  library   exposed-modules:       Haskoin.Node-  other-modules:+      Haskoin.Node.Address       Haskoin.Node.Chain+      Haskoin.Node.Chain.Database+      Haskoin.Node.Chain.Sync       Haskoin.Node.Logger       Haskoin.Node.Loop       Haskoin.Node.Peer+      Haskoin.Node.Peer.Lock+      Haskoin.Node.Peer.Request       Haskoin.Node.PeerMgr-      Paths_haskoin_node+      Haskoin.Node.PeerMgr.Addresses+      Haskoin.Node.PeerMgr.Handshake+      Haskoin.Node.PeerMgr.Lifecycle   hs-source-dirs:       src+  default-extensions:+      DuplicateRecordFields+      MultiWayIf+      NoFieldSelectors+      OverloadedRecordDot+      OverloadedStrings+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -Wunused-packages   build-depends:-      async-    , base >=4.9 && <5+      async ==2.2.*+    , base >=4.20 && <5+    , bytestring >=0.11 && <0.13+    , cereal ==0.5.*+    , conduit ==1.3.*+    , conduit-extra ==1.3.*+    , containers >=0.6 && <0.8+    , fast-logger ==3.2.*+    , haskoin-core >=1.6.1 && <1.7+    , network >=3.1 && <3.3+    , nqe >=0.6.9 && <0.7+    , random >=1.2 && <1.4+    , rocksdb-haskell-jprupp >=2.6.1 && <2.7+    , rocksdb-query >=0.6.1 && <0.7+    , stm ==2.5.*+    , string-conversions ==0.4.*+    , text >=2.0 && <2.2+    , time >=1.12 && <1.15+    , transformers ==0.6.*+  default-language: GHC2024++test-suite regtest+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      Regtest.Podman+      Haskoin.Node.Test+      Paths_haskoin_node+  hs-source-dirs:+      test/regtest+      test/common+  default-extensions:+      DuplicateRecordFields+      MultiWayIf+      NoFieldSelectors+      OverloadedRecordDot+      OverloadedStrings+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -Wunused-packages -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      HUnit+    , base >=4.20 && <5     , bytestring     , cereal     , conduit-    , conduit-extra-    , containers     , data-default-    , fast-logger-    , hashable-    , haskoin-core >=1.6.0-    , mtl+    , directory+    , haskoin-core+    , haskoin-node+    , hspec >=2.10     , network-    , nqe >=0.6.3+    , nqe+    , process     , random-    , rocksdb-haskell-jprupp >=2.3.0-    , rocksdb-query >=0.6.0-    , stm-    , string-conversions-    , text+    , rocksdb-haskell-jprupp     , time-    , transformers     , unliftio-    , unordered-containers-  default-language: Haskell2010+  default-language: GHC2024  test-suite spec   type: exitcode-stdio-1.0   main-is: Spec.hs   other-modules:+      Haskoin.Node.AddressSpec+      Haskoin.Node.Chain.SyncSpec+      Haskoin.Node.ChainSpec+      Haskoin.Node.Peer.LockSpec+      Haskoin.Node.Peer.RequestSpec+      Haskoin.Node.PeerMgr.AddressesSpec+      Haskoin.Node.PeerMgr.HandshakeSpec+      Haskoin.Node.PeerMgr.LifecycleSpec+      Haskoin.Node.PeerMgrSpec+      Haskoin.Node.PeerSpec       Haskoin.NodeSpec+      Haskoin.Node.Test       Paths_haskoin_node   hs-source-dirs:-      test+      test/unit+      test/common+  default-extensions:+      DuplicateRecordFields+      MultiWayIf+      NoFieldSelectors+      OverloadedRecordDot+      OverloadedStrings+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -Wunused-packages -threaded -rtsopts -with-rtsopts=-N   build-depends:       HUnit+    , QuickCheck >=2.14     , async-    , base >=4.9 && <5+    , base >=4.20 && <5     , base64     , bytestring     , cereal     , conduit-    , conduit-extra-    , containers     , data-default-    , fast-logger-    , hashable-    , haskoin-core >=1.6.0+    , haskoin-core     , haskoin-node-    , hspec-    , mtl+    , hspec >=2.10     , network-    , nqe >=0.6.3+    , nqe     , random-    , rocksdb-haskell-jprupp >=2.3.0-    , rocksdb-query >=0.6.0-    , safe-    , stm-    , string-conversions+    , rocksdb-haskell-jprupp     , text     , time-    , transformers     , unliftio-    , unordered-containers-  default-language: Haskell2010-  build-tool-depends: hspec-discover:hspec-discover+  default-language: GHC2024+  build-tool-depends:+      hspec-discover:hspec-discover

@@ -1,13 +1,3 @@-{-# LANGUAGE DuplicateRecordFields #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE NoFieldSelectors #-}- -- | -- Module      : Haskoin.Node -- Copyright   : (c) Haskoin Developers@@ -32,7 +22,9 @@ --   too long. -- -- * The chain ("Haskoin.Node.Chain") downloads block headers from peers,---   connects them to the best chain, and stores them in the database.+--   connects them to the best chain, and stores them in the database. It+--   follows the announcements of new blocks, whether they come as headers or+--   as inventory, and polls peers when it has heard nothing for a while. -- -- Everything that happens is published as a 'NodeEvent' to the 'Publisher' -- given in 'NodeConfig'. Subscribe to it to learn about new best blocks, peers@@ -66,6 +58,13 @@ --   by messages, which keeps the concurrency easy to reason about and lets a --   failing peer die without disturbing the rest. --+-- * __Plain 'IO'.__ Every function runs in 'IO', without monad transformers+--   or @MonadIO@ and @MonadUnliftIO@ constraints. The types say exactly what+--   runs, GHC has nothing to specialise, and the long-running loops of the node+--   cannot leak memory through an unspecialised dictionary (see+--   "Haskoin.Node.Loop"). Programs that work in their own monad call the node+--   with @liftIO@.+-- -- * __Trust nothing from the network.__ Peers are strangers. Every message is --   checked against size limits before it is read, invalid input disconnects --   the peer, and a peer cannot make the node allocate memory faster than it@@ -114,33 +113,33 @@     Node (..),     withNode,     withConnection,+     -- * Peers     module Haskoin.Node.Peer,+     -- * Peer manager     module Haskoin.Node.PeerMgr,+     -- * Chain     module Haskoin.Node.Chain,+     -- * Logging     module Haskoin.Node.Logger,   ) where -import Control.Monad.Cont (ContT (..), MonadCont (callCC), cont, runCont, runContT)-import Control.Monad.Trans (lift)-import Data.Conduit.Network-import Data.String.Conversions (cs)+import Control.Concurrent.Async (link, withAsync) import Data.Time.Clock (NominalDiffTime) import Database.RocksDB (ColumnFamily, DB) import Haskoin+import Haskoin.Node.Address (withConnection) import Haskoin.Node.Chain import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever) import Haskoin.Node.Peer import Haskoin.Node.PeerMgr import NQE-import Network.Socket-import Text.Read (readMaybe)-import UnliftIO+import Network.Socket (SockAddr)  -- | General node configuration. Passed to 'withNode'. --@@ -192,39 +191,13 @@     ChainEvent !ChainEvent   | -- | a peer connected, disconnected or sent a message     PeerEvent !PeerEvent-  deriving (Eq)---- | Open a TCP connection to a peer address and pass its conduits to a--- function. The connection is closed when the function returns. Calls 'error'--- if the address cannot be converted into a host and port.-withConnection :: SockAddr -> WithConnection-withConnection na f =-  fromSockAddr na >>= \case-    Nothing -> error $ "Invalid address " ++ show na-    Just cset ->-      runTCPClient cset $ \ad ->-        f (Conduits (appSource ad) (appSink ad))---- | Turn a socket address into client settings for a TCP connection, or--- 'Nothing' if it cannot be converted.-fromSockAddr :: (MonadUnliftIO m) => SockAddr -> m (Maybe ClientSettings)-fromSockAddr sa = go `catch` e-  where-    go = do-      (maybe_host, maybe_port) <- liftIO $ getNameInfo flags True True sa-      return $-        clientSettings-          <$> (readMaybe =<< maybe_port)-          <*> (cs <$> maybe_host)-    flags = [NI_NUMERICHOST, NI_NUMERICSERV]-    e :: (Monad m) => SomeException -> m (Maybe a)-    e _ = return Nothing+  deriving stock (Eq)  -- | Forward events from the chain to the node publisher, telling the peer -- manager about every new best block height so that it can tell peers apart -- that are behind. chainEvents ::-  (MonadIO m) => PeerMgr -> Inbox ChainEvent -> Publisher NodeEvent -> m ()+  PeerMgr -> Inbox ChainEvent -> Publisher NodeEvent -> IO () chainEvents mgr input output = loopForever $ do   event <- receive input   case event of@@ -235,11 +208,13 @@ -- | Route events from peers to the peer manager and chain. A message from a -- peer is handled to completion before the peer may send another one past its -- small allowance of pending messages (see 'maxPendingMessages'), so that the--- node's queues cannot grow faster than it can work. Address and header--- messages are only taken from peers that completed the handshake.+-- node's queues cannot grow faster than it can work. Address, header and+-- inventory messages are only taken from peers that are online: the handshake+-- is complete and the peer is not being disconnected.+-- Inventory with blocks goes to the chain, which asks for headers if it does+-- not know them. peerEvents ::-  (MonadIO m) =>-  Chain -> PeerMgr -> Inbox PeerEvent -> Publisher NodeEvent -> m ()+  Chain -> PeerMgr -> Inbox PeerEvent -> Publisher NodeEvent -> IO () peerEvents ch mgr input output = loopForever $ do   event <- receive input   case event of@@ -247,8 +222,8 @@       chainPeerConnected ch p     PeerDisconnected p ->       chainPeerDisconnected ch p-    PeerMessage p msg -> do-      case msg of+    PeerMessage p m -> do+      case m of         MVersion v ->           peerMgrVersion mgr p v >> peerMgrSync mgr         MVerAck ->@@ -261,6 +236,9 @@           whenOnline p $ peerMgrAddrs mgr p (map snd ns) >> peerMgrSync mgr         MHeaders (Headers hs) ->           whenOnline p $ chainHeaders ch p (map fst hs) >> chainSync ch+        MInv (Inv ivs) -> case [BlockHash iv.invHash | iv <- ivs, isBlock iv] of+          [] -> return ()+          bhs -> whenOnline p $ chainBlocksAnnounced ch p bhs >> chainSync ch         _ -> return ()       ticklePeer mgr p       peerMessageDone p@@ -268,8 +246,9 @@   where     whenOnline p act =       getOnlinePeer mgr p >>= \case-        Just OnlinePeer {online = True} -> act+        Just o | o.online -> act         _ -> return ()+    isBlock iv = iv.invType `elem` [InvBlock, InvWitnessBlock]  -- | Launch node process in the foreground. --@@ -277,18 +256,39 @@ -- handle to the given action, and stop everything when the action returns. If -- any component fails, the exception is re-thrown in the calling thread. withNode ::-  (MonadUnliftIO m) => NodeConfig -> (Node -> m a) -> m a-withNode NodeConfig {..} action = flip runContT return $ do-  peerPub <- ContT withPublisher-  peerSub <- ContT (withSubscription peerPub)-  chainPub <- ContT withPublisher-  chainSub <- ContT (withSubscription chainPub)-  let peerMgrCfg =-        PeerMgrConfig {pub = peerPub, logger = withSource "PeerMgr" logger, ..}-  let chainCfg =-        ChainConfig {pub = chainPub, logger = withSource "Chain" logger, ..}-  chain <- ContT (withChain chainCfg)-  peerMgr <- ContT $ withPeerMgr peerMgrCfg-  link =<< ContT (withAsync $ chainEvents peerMgr chainSub pub)-  link =<< ContT (withAsync $ peerEvents chain peerMgr peerSub pub)-  lift $ action Node {..}+  NodeConfig -> (Node -> IO a) -> IO a+withNode cfg action =+  withPublisher $ \peerPub ->+    withSubscription peerPub $ \peerSub ->+      withPublisher $ \chainPub ->+        withSubscription chainPub $ \chainSub ->+          withChain (chainConfig chainPub) $ \chain ->+            withPeerMgr (peerMgrConfig peerPub) $ \peerMgr ->+              withAsync (chainEvents peerMgr chainSub cfg.pub) $ \a1 ->+                withAsync (peerEvents chain peerMgr peerSub cfg.pub) $ \a2 -> do+                  link a1+                  link a2+                  action Node {peerMgr = peerMgr, chain = chain}+  where+    chainConfig pub =+      ChainConfig+        { db = cfg.db,+          cf = cfg.cf,+          net = cfg.net,+          pub = pub,+          timeout = cfg.timeout,+          logger = withSource "Chain" cfg.logger+        }+    peerMgrConfig pub =+      PeerMgrConfig+        { maxPeers = cfg.maxPeers,+          peers = cfg.peers,+          discover = cfg.discover,+          address = cfg.address,+          net = cfg.net,+          pub = pub,+          timeout = cfg.timeout,+          maxPeerLife = cfg.maxPeerLife,+          connect = cfg.connect,+          logger = withSource "PeerMgr" cfg.logger+        }

@@ -0,0 +1,157 @@+-- |+-- Module      : Haskoin.Node.Address+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Peer addresses: parsing them, deciding which relayed addresses are worth+-- connecting to, and opening TCP connections.+--+-- "Haskoin.Node.PeerMgr" re-exports 'toHostService', 'toSockAddr' and+-- 'isRoutable', and "Haskoin.Node" re-exports 'withConnection', where they+-- used to be defined. They live here because none of them depends on the+-- peer manager or the node: they are about addresses and sockets only.+module Haskoin.Node.Address+  ( toHostService,+    toSockAddr,+    isRoutable,+    withConnection,+  )+where++import Control.Applicative ((<|>))+import Control.Arrow (second)+import Control.Exception (IOException, catch)+import Data.Bits ((.&.))+import Data.Conduit.Network+  ( ClientSettings,+    appSink,+    appSource,+    clientSettings,+    runTCPClient,+  )+import Data.List (elemIndex)+import Data.String.Conversions (cs)+import Data.Word (Word8)+import Haskoin (Network (..))+import Haskoin.Node.Peer (Conduits (..), WithConnection)+import Network.Socket+  ( AddrInfo (..),+    NameInfoFlag (..),+    SockAddr (..),+    getAddrInfo,+    getNameInfo,+    hostAddress6ToTuple,+    hostAddressToTuple,+  )+import Text.Read (readMaybe)++-- | Split a peer address into host and port. Accepts @host@, @host:port@,+-- @[ipv6]@ and @[ipv6]:port@. Either part is 'Nothing' if missing.+toHostService :: String -> (Maybe String, Maybe String)+toHostService str =+  let host = case m6 of+        Just (x, _) -> Just x+        Nothing -> case takeWhile (/= ':') str of+          [] -> Nothing+          xs -> Just xs+      srv = case m6 of+        Just (_, y) -> s y+        Nothing -> s str+      s xs =+        case dropWhile (/= ':') xs of+          [] -> Nothing+          _ : ys -> Just ys+      m6 = case str of+        (x : xs)+          | x == '[' -> do+              i <- elemIndex ']' xs+              return $ second (drop 1) $ splitAt i xs+          | x == ':' -> do+              return (str, "")+        _ -> Nothing+   in (host, srv)++-- | Resolve a peer address given as in 'toHostService' into socket addresses,+-- using the default port of the network if none is given. Returns an empty list+-- if it cannot be resolved.+--+-- Only resolver failures, which are 'IOException's, are caught. Catching+-- every exception would also swallow the asynchronous exception that cancels+-- the thread doing the lookup, and the node could then never shut down.+toSockAddr :: Network -> String -> IO [SockAddr]+toSockAddr net str =+  resolve `catch` \(_ :: IOException) -> return []+  where+    resolve = map (.addrAddress) <$> getAddrInfo Nothing host srv+    (host, srv) =+      second (<|> Just (show net.defaultPort)) (toHostService str)++-- | Whether an address that a peer relayed is worth connecting to: a public+-- unicast address with a port. Loopback, private, link-local, shared,+-- documentation, multicast and reserved ranges are refused, so that a peer+-- cannot make the node connect to services on its own host or network.+-- Static peers in the configuration are not checked.+isRoutable :: SockAddr -> Bool+isRoutable (SockAddrInet port h) =+  port /= 0 && routable4 (hostAddressToTuple h)+isRoutable (SockAddrInet6 port _ h _) =+  port /= 0 && case hostAddress6ToTuple h of+    (0, 0, 0, 0, 0, 0xffff, hi, lo) ->+      routable4+        ( fromIntegral (hi `div` 256),+          fromIntegral (hi `mod` 256),+          fromIntegral (lo `div` 256),+          fromIntegral (lo `mod` 256)+        )+    (a, b, _, _, _, _, _, _)+      | a == 0 -> False -- unspecified, loopback, IPv4-compatible+      | a .&. 0xfe00 == 0xfc00 -> False -- unique local+      | a .&. 0xffc0 == 0xfe80 -> False -- link-local+      | a .&. 0xffc0 == 0xfec0 -> False -- site-local+      | a .&. 0xff00 == 0xff00 -> False -- multicast+      | a == 0x2001 && b == 0x0db8 -> False -- documentation+      | otherwise -> True+isRoutable _ = False++-- | Whether an IPv4 address is public unicast.+routable4 :: (Word8, Word8, Word8, Word8) -> Bool+routable4 (a, b, c, _)+  | a == 0 = False -- this network+  | a == 10 = False -- private+  | a == 100 && b .&. 0xc0 == 64 = False -- shared address space+  | a == 127 = False -- loopback+  | a == 169 && b == 254 = False -- link-local+  | a == 172 && b .&. 0xf0 == 16 = False -- private+  | a == 192 && b == 0 && c == 0 = False -- protocol assignments+  | a == 192 && b == 0 && c == 2 = False -- documentation+  | a == 192 && b == 168 = False -- private+  | a == 198 && b .&. 0xfe == 18 = False -- benchmarking+  | a == 198 && b == 51 && c == 100 = False -- documentation+  | a == 203 && b == 0 && c == 113 = False -- documentation+  | a >= 224 = False -- multicast and reserved+  | otherwise = True++-- | Open a TCP connection to a peer address and pass its conduits to a+-- function. The connection is closed when the function returns. Calls 'error'+-- if the address cannot be converted into a host and port.+withConnection :: SockAddr -> WithConnection+withConnection sa f =+  fromSockAddr sa >>= \case+    Nothing -> error $ "Invalid address " <> show sa+    Just settings ->+      runTCPClient settings $ \ad ->+        f Conduits {inboundConduit = appSource ad, outboundConduit = appSink ad}++-- | Turn a socket address into client settings for a TCP connection, or+-- 'Nothing' if it cannot be converted.+fromSockAddr :: SockAddr -> IO (Maybe ClientSettings)+fromSockAddr sa =+  resolve `catch` \(_ :: IOException) -> return Nothing+  where+    flags = [NI_NUMERICHOST, NI_NUMERICSERV]+    resolve = do+      (host, port) <- (getNameInfo flags True True sa)+      return $ clientSettings <$> (readMaybe =<< port) <*> (cs <$> host)

@@ -1,20 +1,3 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DuplicateRecordFields #-}-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE ImportQualifiedPost #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE MultiWayIf #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE NoFieldSelectors #-}-{-# OPTIONS_GHC -fno-warn-orphans #-}- -- | -- Module      : Haskoin.Node.Chain -- Copyright   : (c) Haskoin Developers@@ -34,7 +17,9 @@ -- -- After a peer has delivered all its headers, it is asked to announce new -- blocks with headers messages, which are fed back to the chain process by--- "Haskoin.Node".+-- "Haskoin.Node". Peers fall back to inventory messages when they cannot+-- announce with headers, and "Haskoin.Node" feeds those to+-- 'chainBlocksAnnounced'. -- -- = Why it works this way --@@ -42,8 +27,21 @@ --   peers for the same range at once would only download it several times. --   The chain syncs from one peer and moves to the next when it is done or --   times out, which also means every connected peer eventually gets a chance---   to show a better chain.+--   to show a better chain. "Haskoin.Node.Chain.Sync" decides whose turn it+--   is. --+-- * __Every hint of unknown blocks is followed.__ A peer announces a new block+--   with a headers message only while it believes the node has the parent.+--   When many blocks are found at once, or the node missed an announcement,+--   the peer sends an inventory message instead, or headers that do not+--   connect. Either way the node asks that peer for headers. Ignoring these+--   left the chain stuck at an old height until the peer reconnected.+--+-- * __Quiet means poll.__ If no headers are requested or received for+--   'timeout' seconds, every connected peer is asked for headers again. This+--   costs one small request per peer per quiet period, and puts a bound on how+--   long any unforeseen stall can last.+-- -- * __Most work wins.__ The best chain is the one with the most accumulated --   proof of work, not the longest or the first seen, as in every Bitcoin --   implementation.@@ -55,24 +53,14 @@ --   use 'ChainSynced' to start work that only makes sense at the tip, so a false --   signal would be costly. ----- * __Headers are stored, not kept in memory.__ Storing them in RocksDB lets a---   restart resume where it stopped instead of downloading the whole chain---   again.------ = Database------ Headers are stored as 'BlockNode' values in the configured column family, or--- the default one, under keys starting with byte @0x90@. The best block is--- stored under key @0x91@ and a data version under key @0x92@. If the data--- version does not match, keys @0x90@ and @0x91@ are deleted when the chain--- starts. Applications that share the database must not use these prefixes.+-- * __Headers are stored, not kept in memory.__ "Haskoin.Node.Chain.Database"+--   describes how, and which key prefixes it uses in the database. module Haskoin.Node.Chain   ( ChainConfig (..),     ChainEvent (..),     Chain,-    ChainT,-    runChainT,     withChain,+    chainRunHeaders,     chainGetBlock,     chainGetBest,     chainGetAncestor,@@ -83,33 +71,32 @@     chainIsSynced,     chainBlockMain,     chainHeaders,+    chainBlocksAnnounced,     chainSync,   ) where -import Control.Monad (forM_, guard, when)-import Control.Monad.Trans.Reader-import Data.ByteString qualified as B+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (link, withAsync)+import Control.Concurrent.STM+import Control.Monad (filterM, unless, void, when) import Data.Function (on)-import Data.List (delete, nub)-import Data.Maybe (isJust, isNothing)-import Data.Serialize+import Data.Functor ((<&>))+import Data.Maybe (isNothing) import Data.String.Conversions (cs)+import Data.Text (Text) import Data.Time.Clock-import Data.Time.Clock.POSIX-import Data.Word (Word32)+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds) import Database.RocksDB (ColumnFamily, DB)-import Database.RocksDB qualified as R-import Database.RocksDB.Query-import Haskoin hiding (Key)+import Haskoin+import Haskoin.Node.Chain.Database (ChainDB (..), initChainDB, runHeaders)+import Haskoin.Node.Chain.Sync import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever) import Haskoin.Node.Peer import Haskoin.Node.PeerMgr (myVersion) import NQE import System.Random (randomRIO)-import UnliftIO-import UnliftIO.Concurrent (threadDelay)  -- | Configuration for chain syncing process. data ChainConfig = ChainConfig@@ -122,7 +109,8 @@     -- | send header chain events here     pub :: !(Publisher ChainEvent),     -- | seconds to wait for headers from the syncing peer before it is-    -- disconnected+    -- disconnected; also the quiet period after which all peers are asked for+    -- headers again     timeout :: !NominalDiffTime,     -- | log messages here     logger :: !Logger@@ -132,11 +120,13 @@ data ChainMessage   = -- | a peer sent headers     ChainHeaders !Peer ![BlockHeader]+  | -- | a peer announced blocks in an inventory message+    ChainBlocksAnnounced !Peer ![BlockHash]   | -- | a peer completed the handshake     ChainPeerConnected !Peer   | -- | a peer disconnected     ChainPeerDisconnected !Peer-  | -- | periodic tick to check the syncing peer for timeouts+  | -- | periodic tick to check for timeouts and quiet periods     ChainPing   | -- | reply once every previous message has been handled     ChainSyncMsg !(Listen ())@@ -148,7 +138,7 @@   | -- | chain is in sync with the network; published only once, with the best     -- block at that time     ChainSynced !BlockNode-  deriving (Eq, Show)+  deriving stock (Eq, Show)  -- | Handle to a running chain process. Obtain one with 'withChain', or from -- the 'Haskoin.Node.Node' returned by 'Haskoin.Node.withNode'.@@ -158,113 +148,36 @@     -- | mailbox of the chain process     mailbox :: !(Mailbox ChainMessage),     -- | syncing state-    state :: !(TVar ChainState)+    state :: !(TVar ChainState),+    -- | reserves the peer that headers are requested from+    owner :: !Owner   }  instance Eq Chain where   (==) = (==) `on` (.mailbox) --- | Database key for version.-data ChainDataVersionKey = ChainDataVersionKey-  deriving (Eq, Ord, Show)--instance Key ChainDataVersionKey--instance KeyValue ChainDataVersionKey Word32--instance Serialize ChainDataVersionKey where-  get = do-    guard . (== 0x92) =<< getWord8-    return ChainDataVersionKey-  put ChainDataVersionKey = putWord8 0x92---- | State of the sync against one peer.-data ChainSync = ChainSync-  { -- | peer being synced against-    peer :: !Peer,-    -- | when it last sent headers, or when syncing started-    timestamp :: !UTCTime,-    -- | last header it sent, from which to continue syncing-    best :: !(Maybe BlockNode)-  }- -- | Mutable state for the header chain process. data ChainState = ChainState-  { -- | peer to sync against and time of last received message-    syncing :: !(Maybe ChainSync),-    -- | queue of peers to sync against-    peers :: ![Peer],+  { -- | which peers to ask for headers+    sync :: !(SyncState Peer),     -- | has the header chain ever been considered synced?     beenInSync :: !Bool   } --- | Key for block header in database.-newtype BlockHeaderKey = BlockHeaderKey BlockHash deriving (Eq, Show)--instance Serialize BlockHeaderKey where-  get = do-    guard . (== 0x90) =<< getWord8-    BlockHeaderKey <$> get-  put (BlockHeaderKey bh) = do-    putWord8 0x90-    put bh---- | Key for best block in database.-data BestBlockKey = BestBlockKey deriving (Eq, Show)--instance KeyValue BlockHeaderKey BlockNode--instance KeyValue BestBlockKey BlockNode--instance Serialize BestBlockKey where-  get = do-    guard . (== 0x91) =<< getWord8-    return BestBlockKey-  put BestBlockKey = putWord8 0x91---- | Monad transformer with access to a 'Chain'. It has a 'BlockHeaders'--- instance, so the header functions of haskoin-core, such as 'getAncestor' and--- 'splitPoint', can run in it against the database of the chain.-type ChainT m = ReaderT Chain m---- | Run a 'ChainT' action against a chain.-runChainT :: ChainT m a -> Chain -> m a-runChainT = runReaderT+-- | Where the headers of a configuration are stored.+configDB :: ChainConfig -> ChainDB+configDB cfg = ChainDB {db = cfg.db, cf = cfg.cf} --- | Read and write headers in the database of a chain configuration. Orphan--- instance.-instance (MonadIO m) => BlockHeaders (ReaderT ChainConfig m) where-  addBlockHeader bn = ReaderT $ \ChainConfig {db, cf} -> liftIO $ do-    case cf of-      Nothing -> insert db (BlockHeaderKey h) bn-      Just cf' -> insertCF db cf' (BlockHeaderKey h) bn-    where-      h = headerHash bn.header-  getBlockHeader bh = ReaderT $ \ChainConfig {db, cf} -> liftIO $ do-    retrieveCommon db cf (BlockHeaderKey bh)-  getBestBlockHeader = ReaderT $ \ChainConfig {db, cf} -> liftIO $ do-    retrieveCommon db cf BestBlockKey >>= \case-      Nothing -> error "Could not get best block from database"-      Just b -> return b-  setBestBlockHeader bn = ReaderT $ \ChainConfig {db, cf} -> liftIO $ do-    case cf of-      Nothing -> insert db BestBlockKey bn-      Just cf' -> insertCF db cf' BestBlockKey bn-  addBlockHeaders bns = ReaderT $ \ChainConfig {db, cf} -> liftIO $ do-    writeBatch db (map (f cf) bns)-    where-      h bn = headerHash bn.header-      f cf bn = case cf of-        Nothing -> insertOp (BlockHeaderKey (h bn)) bn-        Just cf' -> insertOpCF cf' (BlockHeaderKey (h bn)) bn+-- | Run a header function of haskoin-core against the database of a chain.+withDB :: Chain -> (forall n. (BlockHeaders n) => n a) -> IO a+withDB ch = runHeaders (configDB ch.config) --- | Read and write headers in the database of a chain. Orphan instance.-instance (MonadIO m) => BlockHeaders (ChainT m) where-  addBlockHeader bn = withReaderT (.config) (addBlockHeader bn)-  getBlockHeader bh = withReaderT (.config) (getBlockHeader bh)-  getBestBlockHeader = withReaderT (.config) getBestBlockHeader-  setBestBlockHeader bn = withReaderT (.config) (setBestBlockHeader bn)-  addBlockHeaders bns = withReaderT (.config) (addBlockHeaders bns)+-- | Run header functions of haskoin-core, such as 'getAncestor' or+-- 'blockLocator', against the headers of a chain. The functions that start+-- with @chain@ cover the common queries; this is for the rest.+chainRunHeaders ::+  Chain -> (forall n. (BlockHeaders n) => n a) -> IO a+chainRunHeaders = withDB  -- | Start the chain process and pass its handle to the given action. The process -- is stopped when the action returns, and its exceptions are re-thrown in the@@ -274,462 +187,333 @@ -- seeded with the genesis block of the network if it has no best block. The -- current best block is published as a 'ChainBestBlock' event on start. withChain ::-  (MonadUnliftIO m) => ChainConfig -> (Chain -> m a) -> m a+  ChainConfig -> (Chain -> IO a) -> IO a withChain cfg action = do   (inbox, mailbox) <- newMailbox-  st <--    newTVarIO-      ChainState-        { syncing = Nothing,-          beenInSync = False,-          peers = []-        }-  let ch = Chain {config = cfg, mailbox = mailbox, state = st}-  initChainDB cfg-  withAsync (main_loop ch inbox) $ \a ->+  now <- getCurrentTime+  st <- newTVarIO ChainState {sync = emptySync now, beenInSync = False}+  owner <- newOwner "Chain"+  let ch = Chain {config = cfg, mailbox = mailbox, state = st, owner = owner}+  initChainDB cfg.net (configDB cfg)+  withAsync (withClock cfg.timeout mailbox (run ch inbox)) $ \a ->     link a >> action ch   where-    main_loop ch inbox =-      withSyncLoop ch.mailbox (run ch inbox)     run ch inbox = do-      runReaderT getBestBlockHeader cfg-        >>= chainEvent cfg . ChainBestBlock+      withDB ch getBestBlockHeader >>= chainEvent ch . ChainBestBlock       loopForever $ do-        logDebug ch.config.logger "Awaiting event..."-        msg <- receive inbox-        chainConfigMessage ch msg+        logDebug cfg.logger "Awaiting event..."+        receive inbox >>= chainMessage ch  -- | Log and publish a chain event.-chainEvent :: (MonadIO m) => ChainConfig -> ChainEvent -> m ()-chainEvent cfg e = do+chainEvent :: Chain -> ChainEvent -> IO ()+chainEvent ch e = do   case e of     ChainBestBlock b ->       logInfo-        cfg.logger+        ch.config.logger         ("Best block header at height " <> cs (show b.height))     ChainSynced b ->       logInfo-        cfg.logger+        ch.config.logger         ("Headers in sync at height " <> cs (show b.height))-  publish e cfg.pub+  publish e ch.config.pub --- | Import headers sent by a peer. Disconnect the peer if they do not connect.--- Publish a new best block if it changed. If the peer has no more headers, move--- on to the next peer and check whether the chain is now in sync; otherwise ask--- it for more.-processHeaders :: (MonadIO m) => Chain -> Peer -> [BlockHeader] -> m ()-processHeaders ch p hs = do-  let len = length hs+-- | Handle a message sent to the chain process.+chainMessage :: Chain -> ChainMessage -> IO ()+chainMessage ch = \case+  ChainPeerConnected p -> do+    logDebug ch.config.logger ("Connected peer " <> p.label)+    update ch (peerConnected p)+    syncNext ch+  ChainPeerDisconnected p -> do+    logDebug ch.config.logger ("Disconnected peer " <> p.label)+    releaseIfSyncing ch p+    update ch (peerDisconnected p)+    syncNext ch+  ChainHeaders p hs -> whenConnected ch p (processHeaders ch p hs)+  ChainBlocksAnnounced p bhs -> whenConnected ch p (processAnnouncement ch p bhs)+  ChainPing -> checkSync ch+  ChainSyncMsg r -> atomically (r ())++-- | Ignore messages from peers that are not connected, or that are being+-- disconnected for misbehaving.+whenConnected :: Chain -> Peer -> IO () -> IO ()+whenConnected ch p go =+  getStatus ch p >>= \case+    Just Dropped -> ignore "being dropped"+    Just _ -> go+    Nothing -> ignore "not connected"+  where+    ignore why =+      logDebug ch.config.logger ("Ignoring peer " <> why <> ": " <> p.label)++-- | Import headers sent by a peer.+--+-- An empty message from the syncing peer means it has nothing more. Headers+-- that do not connect mean the node missed an announcement, so the peer is+-- asked for headers. Invalid headers disconnect the peer. After a full+-- message of 'maxHeadersSize' headers the peer is asked for more, and after a+-- shorter one the syncing peer is done.+processHeaders :: Chain -> Peer -> [BlockHeader] -> IO ()+processHeaders ch p [] = do+  logDebug ch.config.logger ("No headers from peer " <> p.label)+  syncing <- isSyncing ch p+  when syncing $ finish ch p+processHeaders ch p hs@(h : _) = do   logDebug     ch.config.logger-    ("Processing " <> cs (show len) <> " headers from peer " <> p.label)-  let net = ch.config.net-  now <- liftIO getCurrentTime-  pbest <- runReaderT getBestBlockHeader ch.config-  importHeaders ch now hs >>= \case+    ("Processing " <> cs (show (length hs)) <> " headers from peer " <> p.label)+  withDB ch (getBlockHeader h.prev) >>= \case     Nothing -> do-      logWarn+      logDebug         ch.config.logger-        ("Could not connect headers from peer " <> p.label)-      killPeer p-    Just done -> do-      setLastReceived ch.state p-      best <- runReaderT getBestBlockHeader ch.config-      when (pbest.header /= best.header) $-        chainEvent ch.config (ChainBestBlock best)-      if done-        then do-          MSendHeaders `sendMessage` p-          finishPeer ch p-          syncNewPeer ch-          syncNotif ch-        else syncPeer ch p---- | Start syncing against the next peer in the queue, unless a peer is already being synced.-syncNewPeer :: (MonadIO m) => Chain -> m ()-syncNewPeer ch =-  getSyncingPeer ch.state >>= \case-    Just _ -> return ()-    Nothing ->-      nextPeer ch >>= \case-        Nothing -> return ()-        Just p -> do-          logDebug ch.config.logger ("Syncing against peer " <> p.label)-          syncPeer ch p---- | Publish a 'ChainSynced' event if the chain just became synced.-syncNotif :: (MonadIO m) => Chain -> m ()-syncNotif ch =-  notifySynced ch >>= \case-    False -> return ()-    True ->-      runReaderT getBestBlockHeader ch.config-        >>= chainEvent ch.config . ChainSynced+        ("Headers from peer " <> p.label <> " do not connect")+      update ch (requestSync p)+      syncNext ch+    Just _ ->+      importHeaders ch hs >>= \case+        Left e -> misbehaving ch p ("could not connect headers: " <> cs e)+        Right lst -> do+          syncing <- isSyncing ch p+          let full = length hs >= fromIntegral maxHeadersSize+          if+            | full && syncing -> requestHeaders ch p (Just lst)+            | full -> update ch (requestSync p) >> syncNext ch+            | syncing -> finish ch p+            | otherwise -> syncNotif ch --- | Ask a peer for headers, continuing from the last header it sent, or from the--- best block. Does nothing if another peer is being synced.-syncPeer :: (MonadIO m) => Chain -> Peer -> m ()-syncPeer ch p = do-  t <- liftIO getCurrentTime-  m <--    chainSyncingPeer ch.state >>= \case-      Just ChainSync {peer = s, best = m}-        | p == s -> syncing_me t m-        | otherwise -> return Nothing-      Nothing -> syncing_new t-  forM_ m $ \g -> do+-- | A peer announced blocks in an inventory message. If any is unknown, ask+-- the peer for headers.+processAnnouncement :: Chain -> Peer -> [BlockHash] -> IO ()+processAnnouncement ch p bhs = do+  unknown <- filterM (fmap isNothing . chainGetBlock ch) bhs+  unless (null unknown) $ do     logDebug       ch.config.logger-      ("Requesting headers from peer " <> p.label)-    MGetHeaders g `sendMessage` p-  where-    syncing_new t =-      setSyncingPeer ch p >>= \case-        False -> return Nothing-        True -> do-          logDebug ch.config.logger ("Locked peer " <> p.label)-          h <- runReaderT getBestBlockHeader ch.config-          Just <$> syncHeaders ch t h p-    syncing_me t m = do-      h <- case m of-        Nothing -> runReaderT getBestBlockHeader ch.config-        Just h -> return h-      Just <$> syncHeaders ch t h p+      ( "Peer "+          <> p.label+          <> " announced "+          <> cs (show (length unknown))+          <> " unknown blocks"+      )+    update ch (requestSync p)+    syncNext ch --- | Handle a message sent to the chain process.-chainConfigMessage :: (MonadIO m) => Chain -> ChainMessage -> m ()-chainConfigMessage ch (ChainHeaders p hs) =-  processHeaders ch p hs-chainConfigMessage ch (ChainPeerConnected p) = do-  logDebug ch.config.logger ("Connected peer " <> p.label)-  addPeer ch.state p-  syncNewPeer ch-chainConfigMessage ch (ChainPeerDisconnected p) = do-  logDebug ch.config.logger ("Disconnected peer " <> p.label)-  finishPeer ch p-  syncNewPeer ch-chainConfigMessage _ (ChainSyncMsg r) =-  atomically (r ())-chainConfigMessage ch ChainPing = do-  logDebug ch.config.logger "Internal clock event"-  let to = ch.config.timeout-  now <- liftIO getCurrentTime-  chainSyncingPeer ch.state >>= \case-    Just ChainSync {peer = p, timestamp = t}-      | now `diffUTCTime` t > to -> do-          logWarn-            ch.config.logger-            ("Syncing peer " <> p.label <> " timed out")-          killPeer p+-- | Disconnect the syncing peer if it has not answered in time. If no peer is+-- syncing, start with a queued one, or queue every peer if nothing has+-- happened for a while.+checkSync :: Chain -> IO ()+checkSync ch = do+  now <- getCurrentTime+  s <- (.sync) <$> readTVarIO ch.state+  let quiet t = now `diffUTCTime` t > ch.config.timeout+  case syncingPeer s of+    Just (p, t)+      | quiet t -> misbehaving ch p "timed out"       | otherwise -> return ()-    Nothing -> syncNewPeer ch+    Nothing+      | not (null (queuedPeers s)) -> syncNext ch+      | quiet (lastActivity s) && not (null (connectedPeers s)) -> do+          logDebug ch.config.logger "No recent header activity, polling peers"+          update ch (pollPeers now)+          syncNext ch+      | otherwise -> return () --- | Run a loop in the background that sends a 'ChainPing' to the chain process--- every 2 to 20 seconds.-withSyncLoop :: (MonadUnliftIO m) => Mailbox ChainMessage -> m a -> m a-withSyncLoop mbox mf =-  withAsync go $ \a ->-    link a >> mf+-- | Unless a peer is being synced against already, start with the first queued+-- peer that 'reservePeer' can reserve. Check whether the chain is synced if there+-- is nobody to sync against.+syncNext :: Chain -> IO ()+syncNext ch = do+  s <- (.sync) <$> readTVarIO ch.state+  when (isNothing (syncingPeer s)) $ go (queuedPeers s)   where-    go = loopForever $ do-      delay <- randomRIO (2 * 10 ^ 6, 2 * 10 ^ 7)-      threadDelay delay-      ChainPing `send` mbox---- | Version of the database.-dataVersion :: Word32-dataVersion = 1---- | Initialize header database. If version is different from current, the--- database is purged of conflicting elements first.-initChainDB :: (MonadIO m) => ChainConfig -> m ()-initChainDB cfg@ChainConfig {db, cf, net} = liftIO $ do-  ver <- retrieveCommon db cf ChainDataVersionKey-  when (ver /= Just dataVersion) $ purgeChainDB cfg-  case cf of-    Nothing -> insert db ChainDataVersionKey dataVersion-    Just cf' -> insertCF db cf' ChainDataVersionKey dataVersion-  retrieveCommon db cf BestBlockKey >>= \b ->-    when (isNothing (b :: Maybe BlockNode)) . flip runReaderT cfg $ do-      addBlockHeader (genesisNode net)-      setBestBlockHeader (genesisNode net)+    go [] = syncNotif ch+    go (p : ps) =+      reservePeer ch.owner p >>= \case+        True -> do+          logDebug ch.config.logger ("Syncing against peer " <> p.label)+          now <- getCurrentTime+          update ch (startSync now p)+          requestHeaders ch p Nothing+        False -> do+          logDebug ch.config.logger ("Could not lock peer: " <> p.label)+          go ps --- | Purge database of elements having keys that may conflict with those used in--- this module. Keys are deleted from the column family that holds the chain,--- one at a time as the iterator walks them, so that memory use does not grow--- with the size of the database.-purgeChainDB :: (MonadIO m) => ChainConfig -> m ()-purgeChainDB ChainConfig {db, cf} = liftIO $ do-  with_iter $ \it -> do-    R.iterSeek it (B.singleton 0x90)-    go it+-- | Ask the syncing peer for the headers after the given block, or after the+-- best block.+requestHeaders :: Chain -> Peer -> Maybe BlockNode -> IO ()+requestHeaders ch p after = do+  now <- getCurrentTime+  update ch (refreshSync now p)+  bn <- maybe (withDB ch getBestBlockHeader) return after+  loc <- withDB ch (blockLocator bn)+  logDebug+    ch.config.logger+    ( "Requesting headers after height "+        <> cs (show bn.height)+        <> " from peer "+        <> p.label+    )+  MGetHeaders GetHeaders {version = myVersion, locator = loc, stop = zero}+    `sendMessage` p   where-    with_iter = case cf of-      Nothing -> R.withIter db-      Just cf' -> R.withIterCF db cf'-    del k = case cf of-      Nothing -> R.delete db k-      Just cf' -> R.deleteCF db cf' k-    go it =-      R.iterKey it >>= \case-        Just k-          | B.take 1 k `elem` [B.singleton 0x90, B.singleton 0x91] -> do-              del k-              R.iterNext it-              go it-        _ -> return ()+    zero = "0000000000000000000000000000000000000000000000000000000000000000" --- | Import a bunch of continuous headers. Returns 'True' when the peer that--- delivered them has no more headers to sync. That is the case when it sent--- fewer than 'maxHeadersSize' (2000) headers, the most that one message may--- hold. An empty headers message, which a peer sends when it has nothing--- after the locator, counts as done too.-importHeaders ::-  (MonadIO m) => Chain -> UTCTime -> [BlockHeader] -> m (Maybe Bool)-importHeaders ch now hs =-  connect >>= \case-    Left e -> do-      logError ch.config.logger ("connectBlocks: " <> cs e)-      return Nothing-    Right bns-      | null bns -> return (Just True)-      | otherwise -> do-          bb <- get_last-          atomically . modifyTVar ch.state $ \s ->-            s {syncing = set_best bb <$> s.syncing}-          return (Just (length bns < fromIntegral maxHeadersSize))-  where-    set_best bb ChainSync {..} = ChainSync {best = bb, ..}-    timestamp = floor (utcTimeToPOSIXSeconds now)-    connect = runReaderT (connectBlocks ch.config.net timestamp hs) ch.config-    get_last = runReaderT (getBlockHeader (headerHash (last hs))) ch.config+-- | The syncing peer has no more headers. Ask it to announce new blocks with+-- headers, release it, and move on.+finish :: Chain -> Peer -> IO ()+finish ch p = do+  logDebug ch.config.logger ("Releasing syncing peer " <> p.label)+  MSendHeaders `sendMessage` p+  update ch (finishSync p)+  void (releasePeer ch.owner p)+  syncNext ch --- | Check if best block header is in sync with the rest of the block chain by--- comparing the best block with the current time, verifying that there are no--- peers in the queue to be synced, and no peer is being synced at the moment.--- This function will only return 'True' once. It should be used to decide--- whether to notify other processes that the header chain has been synced. The--- state of the chain will be flipped to synced when this function returns--- 'True'.------ The best block must also have the 'minimumChainWork' of the network.--- Bitcoin Core requires the same before it leaves initial block download. A--- chain with less work is not the real one, however recent its last block--- is. Below that work 'connectBlocks' does not refuse cheap forks either.-notifySynced :: (MonadIO m) => Chain -> m Bool-notifySynced ch = do-  bb <- runReaderT getBestBlockHeader ch.config-  df <- (`diffUTCTime` block_time bb) <$> liftIO getCurrentTime-  atomically $ do-    s <- readTVar ch.state-    if-      | df > 7200 -> return False-      | bb.work < ch.config.net.minimumChainWork -> return False-      | isJust s.syncing -> return False-      | not (null s.peers) -> return False-      | s.beenInSync -> return False-      | otherwise -> do-          writeTVar ch.state s {beenInSync = True}-          return True-  where-    block_time =-      posixSecondsToUTCTime . fromIntegral . (.header.timestamp)+-- | Disconnect a peer that sent invalid headers or timed out, and move on+-- without waiting for its disconnection.+misbehaving :: Chain -> Peer -> Text -> IO ()+misbehaving ch p why = do+  logWarn ch.config.logger ("Disconnecting peer " <> p.label <> ": " <> why)+  releaseIfSyncing ch p+  update ch (dropPeer p)+  killPeer p+  syncNext ch --- | Get next peer to sync against from the queue.-nextPeer :: (MonadIO m) => Chain -> m (Maybe Peer)-nextPeer ch = fmap (.peers) (readTVarIO ch.state) >>= go-  where-    go [] = return Nothing-    go (p : ps) =-      setSyncingPeer ch p >>= \case-        True -> return (Just p)-        False -> go ps+-- | Release a peer reserved for syncing.+releaseIfSyncing :: Chain -> Peer -> IO ()+releaseIfSyncing ch p = do+  syncing <- isSyncing ch p+  when syncing . void $ releasePeer ch.owner p --- | Set a syncing peer and generate a 'GetHeaders' data structure with a block--- locator to send to that peer for syncing.-syncHeaders ::-  (MonadIO m) => Chain -> UTCTime -> BlockNode -> Peer -> m GetHeaders-syncHeaders ch now bb p = do-  atomically $-    modifyTVar ch.state $ \s ->-      s-        { syncing =-            Just-              ChainSync-                { peer = p,-                  timestamp = now,-                  best = Nothing-                },-          peers = delete p s.peers-        }-  loc <- runReaderT (blockLocator bb) ch.config-  return-    GetHeaders-      { version = myVersion,-        locator = loc,-        stop = z-      }-  where-    z = "0000000000000000000000000000000000000000000000000000000000000000"+-- | Apply a transition to the sync state.+update :: Chain -> (SyncState Peer -> SyncState Peer) -> IO ()+update ch f = atomically . modifyTVar ch.state $ \s -> s {sync = f s.sync} --- | Set the time of last received data to now if the given peer is the one--- being synced against. Headers from other peers must not keep a silent--- syncing peer from timing out.-setLastReceived :: (MonadIO m) => TVar ChainState -> Peer -> m ()-setLastReceived st p = do-  now <- liftIO getCurrentTime-  let f c@ChainSync {..}-        | peer == p = ChainSync {timestamp = now, ..}-        | otherwise = c-  atomically (modifyTVar st (\s -> s {syncing = f <$> s.syncing}))+getStatus :: Chain -> Peer -> IO (Maybe PeerStatus)+getStatus ch p = peerStatus p . (.sync) <$> readTVarIO ch.state --- | Add a new peer to the queue of peers to sync against.-addPeer :: (MonadIO m) => TVar ChainState -> Peer -> m ()-addPeer st p = do-  atomically (modifyTVar st (\s -> s {peers = nub (p : s.peers)}))+isSyncing :: Chain -> Peer -> IO Bool+isSyncing ch p =+  getStatus ch p <&> \case+    Just (Syncing _ _) -> True+    _ -> False --- | Get syncing peer if there is one.-getSyncingPeer :: (MonadIO m) => TVar ChainState -> m (Maybe Peer)-getSyncingPeer st =-  readTVarIO st >>= \case-    ChainState {syncing = Just ChainSync {peer}} -> return (Just peer)-    _ -> return Nothing+-- | Validate and store a run of headers, and publish the best block if it+-- changed. Returns the node of the last header.+importHeaders ::+  Chain -> [BlockHeader] -> IO (Either String BlockNode)+importHeaders ch hs = do+  now <- getCurrentTime+  let timestamp = floor (utcTimeToPOSIXSeconds now)+  before <- withDB ch getBestBlockHeader+  withDB ch (connectBlocks ch.config.net timestamp hs) >>= \case+    Left e -> do+      logError ch.config.logger ("connectBlocks: " <> cs e)+      return (Left e)+    Right [] -> return (Left "no headers")+    Right (lst : _) -> do+      update ch (touchSync now)+      after <- withDB ch getBestBlockHeader+      when (before.header /= after.header) $+        chainEvent ch (ChainBestBlock after)+      return (Right lst) --- | Reserve a peer with 'setBusy' and make it the syncing peer. Returns 'False'--- if the peer is busy with something else.-setSyncingPeer :: (MonadIO m) => Chain -> Peer -> m Bool-setSyncingPeer ch p =-  setBusy p >>= \case-    False -> do-      logDebug-        ch.config.logger-        ("Could not lock peer: " <> p.label)-      return False-    True -> do-      logDebug ch.config.logger $-        ("Locked peer: " <> p.label)-      set_it-      return True-  where-    set_it = do-      now <- liftIO getCurrentTime-      atomically $ modifyTVar ch.state $ \s ->-        s-          { syncing =-              Just-                ChainSync-                  { peer = p,-                    timestamp = now,-                    best = Nothing-                  }-          }+-- | Publish a 'ChainSynced' event the first time the chain is synced (see+-- 'isSynced').+syncNotif :: Chain -> IO ()+syncNotif ch = do+  best <- withDB ch getBestBlockHeader+  now <- getCurrentTime+  notify <- atomically $ do+    s <- readTVar ch.state+    if isSynced ch.config.net now best s.sync && not s.beenInSync+      then writeTVar ch.state s {beenInSync = True} >> return True+      else return False+  when notify $ chainEvent ch (ChainSynced best) --- | Remove a peer from the queue of peers to sync and unset the syncing peer if--- it is set to the provided peer.-finishPeer :: (MonadIO m) => Chain -> Peer -> m ()-finishPeer ch p =-  remove_peer >>= \case-    False ->-      logDebug-        ch.config.logger-        ("Removed peer " <> p.label <> " from queue")-    True -> do-      logDebug-        ch.config.logger-        ("Releasing syncing peer " <> p.label)-      setFree p+-- | Run a loop in the background that sends a 'ChainPing' to the chain+-- process at random intervals. They are at most a quarter of the timeout, so+-- timeouts are noticed promptly, and between 0.1 and 20 seconds, so a short+-- timeout does not make the chain spin and a long one does not make it doze.+withClock ::+  NominalDiffTime -> Mailbox ChainMessage -> IO a -> IO a+withClock to mbox go =+  withAsync clock $ \a -> link a >> go   where-    remove_peer =-      atomically $-        readTVar ch.state >>= \s -> case s.syncing of-          Just ChainSync {peer = p'}-            | p == p' -> do-                unset_syncing-                return True-          _ -> do-            remove_from_queue-            return False-    unset_syncing =-      modifyTVar ch.state $ \x ->-        x {syncing = Nothing}-    remove_from_queue =-      modifyTVar ch.state $ \x ->-        x {peers = delete p x.peers}---- | Return syncing peer data.-chainSyncingPeer :: (MonadIO m) => TVar ChainState -> m (Maybe ChainSync)-chainSyncingPeer st = (.syncing) <$> readTVarIO st+    upper :: Int+    upper = max 100000 (min 20000000 (floor (to * 1000000) `div` 4))+    clock = loopForever $ do+      threadDelay =<< randomRIO (upper `div` 10, upper)+      ChainPing `send` mbox  -- | Get a block header from the block chain. Returns headers on side chains -- too; use 'chainBlockMain' to check whether a block is in the main chain.-chainGetBlock :: (MonadIO m) => Chain -> BlockHash -> m (Maybe BlockNode)-chainGetBlock ch bh = runReaderT (getBlockHeader bh) ch.config+chainGetBlock :: Chain -> BlockHash -> IO (Maybe BlockNode)+chainGetBlock ch bh = withDB ch (getBlockHeader bh)  -- | Get best block header from chain process.-chainGetBest :: (MonadIO m) => Chain -> m BlockNode-chainGetBest ch = runReaderT getBestBlockHeader ch.config+chainGetBest :: Chain -> IO BlockNode+chainGetBest ch = withDB ch getBestBlockHeader  -- | Get ancestor of 'BlockNode' at 'BlockHeight' from chain process. Returns -- 'Nothing' if the height is above that of the node. chainGetAncestor ::-  (MonadIO m) => Chain -> BlockHeight -> BlockNode -> m (Maybe BlockNode)-chainGetAncestor ch h bn = runReaderT (getAncestor h bn) ch.config+  Chain -> BlockHeight -> BlockNode -> IO (Maybe BlockNode)+chainGetAncestor ch h bn = withDB ch (getAncestor h bn)  -- | Get parents of 'BlockNode' starting at 'BlockHeight' from chain process. -- The result is in ascending order, from the block at the given height up to -- the parent of the given node, which is not included. chainGetParents ::-  (MonadIO m) => Chain -> BlockHeight -> BlockNode -> m [BlockNode]-chainGetParents ch height top =-  go [] top+  Chain -> BlockHeight -> BlockNode -> IO [BlockNode]+chainGetParents ch height = go []   where     go acc b       | height >= b.height = return acc-      | otherwise = do-          m <- chainGetBlock ch b.header.prev-          case m of+      | otherwise =+          chainGetBlock ch b.header.prev >>= \case             Nothing -> return acc             Just p -> go (p : acc) p  -- | Get last common block from chain process: the most recent block that is an -- ancestor of both nodes. chainGetSplitBlock ::-  (MonadIO m) => Chain -> BlockNode -> BlockNode -> m BlockNode-chainGetSplitBlock ch l r = runReaderT (splitPoint l r) ch.config+  Chain -> BlockNode -> BlockNode -> IO BlockNode+chainGetSplitBlock ch l r = withDB ch (splitPoint l r) --- | Notify chain that a new peer is connected. It is added to the queue of--- peers to sync against.-chainPeerConnected :: (MonadIO m) => Chain -> Peer -> m ()-chainPeerConnected ch p =-  ChainPeerConnected p `send` ch.mailbox+-- | Notify chain that a new peer is connected. It is queued to be asked for+-- headers.+chainPeerConnected :: Chain -> Peer -> IO ()+chainPeerConnected ch p = ChainPeerConnected p `send` ch.mailbox  -- | Notify chain that a peer has disconnected.-chainPeerDisconnected :: (MonadIO m) => Chain -> Peer -> m ()-chainPeerDisconnected ch p =-  ChainPeerDisconnected p `send` ch.mailbox+chainPeerDisconnected :: Chain -> Peer -> IO ()+chainPeerDisconnected ch p = ChainPeerDisconnected p `send` ch.mailbox  -- | Is given 'BlockHash' in the main chain?-chainBlockMain :: (MonadIO m) => Chain -> BlockHash -> m Bool-chainBlockMain ch bh =-  chainGetBest ch >>= \bb ->-    chainGetBlock ch bh >>= \case-      Nothing ->-        return False-      bm@(Just bn) ->-        (== bm) <$> chainGetAncestor ch bn.height bb+chainBlockMain :: Chain -> BlockHash -> IO Bool+chainBlockMain ch bh = do+  best <- chainGetBest ch+  chainGetBlock ch bh >>= \case+    Nothing -> return False+    Just bn -> (== Just bn) <$> chainGetAncestor ch bn.height best  -- | Is chain in sync with network? Once 'True', it stays 'True' even if the -- chain falls behind later.-chainIsSynced :: (MonadIO m) => Chain -> m Bool-chainIsSynced ch =-  (.beenInSync) <$> readTVarIO (ch.state)+chainIsSynced :: Chain -> IO Bool+chainIsSynced ch = (.beenInSync) <$> readTVarIO ch.state  -- | Peer sends a bunch of headers to the chain process. "Haskoin.Node" calls -- this for every headers message received from an online peer.-chainHeaders :: (MonadIO m) => Chain -> Peer -> [BlockHeader] -> m ()-chainHeaders ch p hs =-  ChainHeaders p hs `send` ch.mailbox+chainHeaders :: Chain -> Peer -> [BlockHeader] -> IO ()+chainHeaders ch p hs = ChainHeaders p hs `send` ch.mailbox +-- | Peer announced blocks in an inventory message. If any of them is unknown,+-- the peer is asked for headers. "Haskoin.Node" calls this for every+-- inventory message with blocks received from an online peer.+chainBlocksAnnounced :: Chain -> Peer -> [BlockHash] -> IO ()+chainBlocksAnnounced ch p bhs = ChainBlocksAnnounced p bhs `send` ch.mailbox+ -- | Wait until the chain process has handled every message sent to it before.-chainSync :: (MonadIO m) => Chain -> m ()+chainSync :: Chain -> IO () chainSync ch = ChainSyncMsg `query` ch.mailbox

@@ -0,0 +1,169 @@+-- |+-- Module      : Haskoin.Node.Chain.Database+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Storage of block headers in RocksDB, for "Haskoin.Node.Chain".+--+-- Headers are stored as 'BlockNode' values in the configured column family, or+-- the default one, under keys starting with byte @0x90@. The best block is+-- stored under key @0x91@ and a data version under key @0x92@. If the data+-- version does not match, keys @0x90@ and @0x91@ are deleted when the chain+-- starts. Applications that share the database must not use these prefixes.+--+-- = Why it works this way+--+-- * __Headers are stored, not kept in memory.__ Storing them lets a restart+--   resume where it stopped instead of downloading the whole chain again.+--+-- * __Through haskoin-core's interface.__ The header functions of+--   haskoin-core, such as 'connectBlocks' and 'getAncestor', run in any monad+--   with a 'BlockHeaders' instance. 'runHeaders' runs them against the+--   database, so validation and storage are not reimplemented here. The monad+--   that carries the database is a 'ReaderT', and it stays inside this+--   module: everything else calls 'runHeaders' and works in plain 'IO'.+--+-- * __Separate from syncing.__ This module only knows how headers are laid+--   out on disk. The chain process decides what to store and when.+module Haskoin.Node.Chain.Database+  ( ChainDB (..),+    initChainDB,+    runHeaders,+  )+where++import Control.Monad (guard, unless, when)+import Control.Monad.Trans.Reader (ReaderT (..))+import Data.ByteString qualified as B+import Data.Maybe (isNothing)+import Data.Serialize (Serialize (..), getWord8, putWord8)+import Data.Word (Word32)+import Database.RocksDB (ColumnFamily, DB)+import Database.RocksDB qualified as R+import Database.RocksDB.Query+  ( Key,+    KeyValue,+    insert,+    insertCF,+    insertOp,+    insertOpCF,+    retrieveCommon,+    writeBatch,+  )+import Haskoin hiding (Key)++-- | Where headers are stored: a database and an optional column family.+data ChainDB = ChainDB+  { -- | database handle+    db :: !DB,+    -- | column family, or the default one if 'Nothing'+    cf :: !(Maybe ColumnFamily)+  }++-- | Key of the data version.+data VersionKey = VersionKey+  deriving stock (Eq, Show)++instance Serialize VersionKey where+  get = getWord8 >>= guard . (== 0x92) >> return VersionKey+  put VersionKey = putWord8 0x92++instance Key VersionKey++instance KeyValue VersionKey Word32++-- | Key of a block header.+newtype BlockHeaderKey = BlockHeaderKey BlockHash+  deriving stock (Eq, Show)++instance Serialize BlockHeaderKey where+  get = getWord8 >>= guard . (== 0x90) >> BlockHeaderKey <$> get+  put (BlockHeaderKey bh) = putWord8 0x90 >> put bh++instance KeyValue BlockHeaderKey BlockNode++-- | Key of the best block.+data BestBlockKey = BestBlockKey+  deriving stock (Eq, Show)++instance Serialize BestBlockKey where+  get = getWord8 >>= guard . (== 0x91) >> return BestBlockKey+  put BestBlockKey = putWord8 0x91++instance KeyValue BestBlockKey BlockNode++-- | Run header functions of haskoin-core against the database.+runHeaders ::+  ChainDB -> (forall n. (BlockHeaders n) => n a) -> IO a+runHeaders c action = (runReaderT action c)++-- | Read and write headers in the database.+instance BlockHeaders (ReaderT ChainDB IO) where+  addBlockHeader bn = ReaderT $ \c -> insertCommon c (nodeKey bn) bn+  getBlockHeader bh =+    ReaderT $ \c -> retrieveCommon c.db c.cf (BlockHeaderKey bh)+  getBestBlockHeader = ReaderT $ \c ->+    (retrieveCommon c.db c.cf BestBlockKey) >>= \case+      Nothing -> error "Could not get best block from database"+      Just b -> return b+  setBestBlockHeader bn = ReaderT $ \c -> insertCommon c BestBlockKey bn+  addBlockHeaders bns =+    ReaderT $ \c -> writeBatch c.db $ map (op c) bns+    where+      op c bn = case c.cf of+        Nothing -> insertOp (nodeKey bn) bn+        Just cf -> insertOpCF cf (nodeKey bn) bn++nodeKey :: BlockNode -> BlockHeaderKey+nodeKey bn = BlockHeaderKey (headerHash bn.header)++insertCommon ::+  (KeyValue k v, Serialize k, Serialize v) =>+  ChainDB ->+  k ->+  v ->+  IO ()+insertCommon c k v = case c.cf of+  Nothing -> insert c.db k v+  Just cf -> insertCF c.db cf k v++-- | Version of the data layout.+dataVersion :: Word32+dataVersion = 1++-- | Prepare the database for a network. If its data version is not the+-- current one, headers and best block are deleted first. An empty database+-- gets the genesis block of the network as its best block.+initChainDB :: Network -> ChainDB -> IO ()+initChainDB net c = do+  ver <- retrieveCommon c.db c.cf VersionKey+  unless (ver == Just dataVersion) $ purgeChainDB c+  insertCommon c VersionKey dataVersion+  best <- retrieveCommon c.db c.cf BestBlockKey+  when (isNothing (best :: Maybe BlockNode)) $ runHeaders c $ do+    addBlockHeader (genesisNode net)+    setBestBlockHeader (genesisNode net)++-- | Delete headers and best block from the column family that holds the+-- chain. Keys are deleted one at a time as the iterator walks them, so that+-- memory use does not grow with the size of the database.+purgeChainDB :: ChainDB -> IO ()+purgeChainDB c = withIter $ \it -> do+  R.iterSeek it (B.singleton 0x90)+  let go =+        R.iterKey it >>= \case+          Just k | B.take 1 k `elem` [B.singleton 0x90, B.singleton 0x91] -> do+            case c.cf of+              Nothing -> R.delete c.db k+              Just cf -> R.deleteCF c.db cf k+            R.iterNext it+            go+          _ -> return ()+  go+  where+    withIter = case c.cf of+      Nothing -> R.withIter c.db+      Just cf -> R.withIterCF c.db cf

@@ -0,0 +1,204 @@+-- |+-- Module      : Haskoin.Node.Chain.Sync+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Which peer to ask for headers, as an explicit state machine.+--+-- Every connected peer has a 'PeerStatus'. "Haskoin.Node.Chain" asks one+-- peer at a time for headers. A peer is queued when it connects, and again+-- whenever it tells the node about blocks that the node does not have. If+-- nothing happens for a while, every idle peer is queued again.+--+-- = Why it works this way+--+-- * __Announcements can be missed.__ Peers announce a new block with a+--   headers message only while they believe the node knows its parent, and+--   fall back to an inventory message when many blocks arrive at once. If+--   the node only ever asked for headers when a peer connected, a missed or+--   unexpected announcement would leave it stuck at an old height for good.+--   So every hint of unknown blocks queues the peer, and a quiet period queues+--   every idle peer, which bounds how long any stall can last.+--+-- * __A peer that announced while syncing is asked again.__ Its last reply may+--   have been computed before the new block existed. The 'Syncing' status+--   remembers the announcement so that the peer is queued again when it is+--   done, rather than being forgotten or asked twice at once.+--+-- * __Misbehaving peers stay out.__ A peer that sent invalid headers or timed+--   out is marked 'Dropped' until its disconnection arrives, so nothing can+--   queue it in between.+--+-- * __Pure and polymorphic.__ The rules are plain functions on a value, over+--   any peer type with equality, so they can be tested exhaustively with+--   integers instead of network connections. Transitions that do not apply to+--   the current status leave the state unchanged, which makes them safe to+--   call whenever the matching event happens.+module Haskoin.Node.Chain.Sync+  ( SyncState,+    PeerStatus (..),+    emptySync,+    peerConnected,+    peerDisconnected,+    requestSync,+    startSync,+    refreshSync,+    finishSync,+    dropPeer,+    pollPeers,+    touchSync,+    peerStatus,+    syncingPeer,+    queuedPeers,+    connectedPeers,+    lastActivity,+    isSynced,+    maxTipAge,+  )+where++import Data.Maybe (isNothing, listToMaybe)+import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime)+import Data.Time.Clock.POSIX (posixSecondsToUTCTime)+import Haskoin (BlockHeader (..), BlockNode (..), Network (..))++-- | Sync status of a connected peer.+data PeerStatus+  = -- | Nothing to ask it for.+    Idle+  | -- | Waiting for its turn to be asked for headers.+    Queued+  | -- | Being asked for headers. Holds the time of the last request, and+    -- whether the peer announced more blocks meanwhile, in which case it is+    -- queued again when done.+    Syncing !UTCTime !Bool+  | -- | Misbehaved and is being disconnected. Never asked for headers again.+    Dropped+  deriving stock (Eq, Show)++-- | Connected peers in queue order, with the last time that headers were+-- requested or received.+data SyncState p = SyncState+  { peers :: ![(p, PeerStatus)],+    activity :: !UTCTime+  }+  deriving stock (Eq, Show)++-- | No peers, with the given time as the last activity.+emptySync :: UTCTime -> SyncState p+emptySync = SyncState []++-- | A peer completed the handshake. It is queued to be asked for headers.+peerConnected :: (Eq p) => p -> SyncState p -> SyncState p+peerConnected p s+  | Just _ <- peerStatus p s = s+  | otherwise = s {peers = s.peers <> [(p, Queued)]}++-- | A peer disconnected.+peerDisconnected :: (Eq p) => p -> SyncState p -> SyncState p+peerDisconnected p s = s {peers = filter ((/= p) . fst) s.peers}++-- | The peer told the node about blocks that it does not have.+requestSync :: (Eq p) => p -> SyncState p -> SyncState p+requestSync p s =+  case peerStatus p s of+    Just Idle -> enqueue p s+    Just (Syncing t _) -> setStatus p (Syncing t True) s+    _ -> s++-- | Start asking a queued peer for headers, unless another peer is being+-- asked already.+startSync :: (Eq p) => UTCTime -> p -> SyncState p -> SyncState p+startSync now p s =+  case (syncingPeer s, peerStatus p s) of+    (Nothing, Just Queued) -> setStatus p (Syncing now False) s {activity = now}+    _ -> s++-- | Another request was sent to the peer being synced against.+refreshSync :: (Eq p) => UTCTime -> p -> SyncState p -> SyncState p+refreshSync now p s =+  case peerStatus p s of+    Just (Syncing _ again) -> setStatus p (Syncing now again) s {activity = now}+    _ -> s++-- | The peer being synced against has no more headers to send.+finishSync :: (Eq p) => p -> SyncState p -> SyncState p+finishSync p s =+  case peerStatus p s of+    Just (Syncing _ True) -> enqueue p s+    Just (Syncing _ False) -> setStatus p Idle s+    _ -> s++-- | The peer misbehaved and is being disconnected.+dropPeer :: (Eq p) => p -> SyncState p -> SyncState p+dropPeer p s =+  case peerStatus p s of+    Just _ -> setStatus p Dropped s+    Nothing -> s++-- | Queue every idle peer, and record the time as activity so that polling+-- happens at most once per quiet period.+pollPeers :: UTCTime -> SyncState p -> SyncState p+pollPeers now s = SyncState {peers = map poll s.peers, activity = now}+  where+    poll (p, Idle) = (p, Queued)+    poll x = x++-- | Headers were received at the given time.+touchSync :: UTCTime -> SyncState p -> SyncState p+touchSync now s = s {activity = now}++-- | Status of a peer, or 'Nothing' if it is not connected.+peerStatus :: (Eq p) => p -> SyncState p -> Maybe PeerStatus+peerStatus p s = lookup p s.peers++-- | The peer being synced against, with the time of the last request to it.+syncingPeer :: SyncState p -> Maybe (p, UTCTime)+syncingPeer s = listToMaybe [(p, t) | (p, Syncing t _) <- s.peers]++-- | Queued peers, first in line first.+queuedPeers :: SyncState p -> [p]+queuedPeers s = [p | (p, Queued) <- s.peers]++-- | Connected peers that have not been dropped.+connectedPeers :: SyncState p -> [p]+connectedPeers s = [p | (p, st) <- s.peers, st /= Dropped]++-- | Last time headers were requested or received, or peers were polled.+lastActivity :: SyncState p -> UTCTime+lastActivity s = s.activity++-- | Oldest that the best block may be for the chain to count as synced: two+-- hours, the most that a block timestamp may be ahead of the network's time.+maxTipAge :: NominalDiffTime+maxTipAge = 7200++-- | Whether the header chain counts as synced at the given time: the best+-- block is at most 'maxTipAge' old, it has the 'minimumChainWork' of the+-- network, and no peer is being synced against or waiting to be.+--+-- Recent headers alone do not prove much: a peer can feed a cheap chain with+-- fresh timestamps. Bitcoin Core requires the same work before it leaves+-- initial block download. And while peers are still queued, a better chain may+-- be about to arrive.+isSynced :: Network -> UTCTime -> BlockNode -> SyncState p -> Bool+isSynced net now best s = recent && enough && idle+  where+    tipTime = posixSecondsToUTCTime (fromIntegral best.header.timestamp)+    recent = now `diffUTCTime` tipTime <= maxTipAge+    enough = best.work >= net.minimumChainWork+    idle = isNothing (syncingPeer s) && null (queuedPeers s)++setStatus :: (Eq p) => p -> PeerStatus -> SyncState p -> SyncState p+setStatus p st s = s {peers = map set s.peers}+  where+    set (q, old)+      | q == p = (q, st)+      | otherwise = (q, old)++-- | Put a peer at the end of the queue.+enqueue :: (Eq p) => p -> SyncState p -> SyncState p+enqueue p s = s {peers = filter ((/= p) . fst) s.peers <> [(p, Queued)]}

@@ -1,11 +1,5 @@-{-# LANGUAGE ImportQualifiedPost #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE NoFieldSelectors #-}- -- | Logging to standard error. A 'Logger' is a plain value carried in--- configuration records, so code that logs only needs 'MonadIO'.+-- configuration records, so code that logs only needs 'IO'. -- -- The node used to log through a logging monad transformer. That forced every -- function into that monad, and made polymorphic loops leak memory when GHC did@@ -24,6 +18,7 @@   ) where +import Control.Exception (bracket) import Control.Monad (when) import Data.Text (Text) import Data.Text qualified as T@@ -34,7 +29,6 @@     newFastLogger1,     toLogStr,   )-import UnliftIO (MonadIO, bracket, liftIO)  -- | Severity of a log message, from least to most severe. data LogLevel@@ -74,9 +68,9 @@  -- | Log a message at a level, if the level is at least that of the 'Logger'. -- Lines look like monad-logger's: @[Info#Chain] message@.-logAt :: (MonadIO m) => LogLevel -> Logger -> Text -> m ()+logAt :: LogLevel -> Logger -> Text -> IO () logAt l logger msg =-  when (l >= logger.level) . liftIO . logger.push $+  when (l >= logger.level) . logger.push $     "[" <> toLogStr (drop 5 (show l)) <> tag <> "] " <> toLogStr msg <> "\n"   where     tag@@ -84,7 +78,7 @@       | otherwise = "#" <> toLogStr logger.source  -- | Log at 'LevelDebug', 'LevelInfo', 'LevelWarn' or 'LevelError'.-logDebug, logInfo, logWarn, logError :: (MonadIO m) => Logger -> Text -> m ()+logDebug, logInfo, logWarn, logError :: Logger -> Text -> IO () logDebug = logAt LevelDebug logInfo = logAt LevelInfo logWarn = logAt LevelWarn

@@ -1,16 +1,3 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DuplicateRecordFields #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ImportQualifiedPost #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE NoFieldSelectors #-}- -- | -- Module      : Haskoin.Node.Peer -- Copyright   : (c) Haskoin Developers@@ -37,17 +24,20 @@ --   the payload read, so a peer cannot make the node buffer a huge message it --   would refuse anyway. ----- * Anything that does not parse closes the connection. There is no attempt to---   recover, because a peer that sends garbage once is not worth the risk.+-- * Anything that does not parse closes the connection with a+--   'PeerException' saying why. There is no attempt to recover, because a+--   peer that sends garbage once is not worth the risk. -- -- * Reading pauses when the node falls behind ('maxPendingMessages'), so the --   speed of the node, not the speed of the peer, sets the pace. module Haskoin.Node.Peer   ( PeerConfig (..),     PeerEvent (..),+    PeerException (..),     Conduits (..),     WithConnection,     Peer (..),+    PeerMessage (..),     peer,     wrapPeer,     sendMessage,@@ -56,6 +46,12 @@     getTxs,     getData,     pingPeer,+    Owner,+    newOwner,+    ownerName,+    reservePeer,+    releasePeer,+    peerHolder,     getBusy,     setBusy,     setFree,@@ -65,14 +61,14 @@ where  import Conduit-import Control.Exception+import Control.Concurrent.Async (link, withAsync)+import Control.Concurrent.STM+import Control.Exception (Exception, throwIO) import Control.Monad-import Control.Monad.Trans.Maybe import Data.Bool (bool) import Data.ByteString (ByteString) import Data.ByteString qualified as B import Data.Function (on)-import Data.List (union) import Data.Maybe (isJust) import Data.Serialize (decode, runGet, runPut) import Data.String.Conversions (cs)@@ -81,9 +77,11 @@ import Haskoin import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever)+import Haskoin.Node.Peer.Lock+import Haskoin.Node.Peer.Request import NQE import System.Random-import UnliftIO+import System.Timeout (timeout)  -- | The two directions of a network connection, as conduits. --@@ -124,6 +122,26 @@     PeerMessage !Peer !Message   deriving (Eq) +-- | Why a peer process stopped with an error. The peer manager logs it and+-- forgets the peer.+--+-- These used to be calls to 'error' with a message. A type lets callers and+-- tests tell the reasons apart without parsing text.+data PeerException+  = -- | the peer closed the connection in the middle of a message+    ConnectionClosed+  | -- | the 24-byte message header could not be parsed+    InvalidMessageHeader+  | -- | the message is for another network; holds its magic bytes+    WrongNetwork !Word32+  | -- | the payload is larger than allowed for the command; holds its size+    MessageTooLarge !MessageCommand !Word32+  | -- | the payload could not be parsed; holds the reason+    InvalidPayload !MessageCommand !String+  deriving stock (Show)++instance Exception PeerException+ -- | Mailbox for a peer. Peers are compared by their mailbox, and shown by their -- label. data Peer = Peer@@ -133,8 +151,9 @@     pub :: !(Publisher PeerEvent),     -- | name of the peer in logs, usually its address     label :: !Text,-    -- | whether the peer is reserved for a task; see 'setBusy'-    busy :: !(TVar Bool),+    -- | who has reserved the peer for a long conversation; see+    -- 'reservePeer'+    busy :: !(TVar (Lock Owner)),     -- | messages from this peer that the node has not finished handling     pending :: !(TVar Int)   }@@ -152,11 +171,11 @@   | -- | send a network message to the peer     SendMessage !Message --- | Build a 'Peer' handle from its configuration, its busy flag, its counter of--- pending messages, and the mailbox of its process.+-- | Build a 'Peer' handle from its configuration, its lock (which starts+-- 'Free'), its counter of pending messages, and the mailbox of its process. wrapPeer ::   PeerConfig ->-  TVar Bool ->+  TVar (Lock Owner) ->   TVar Int ->   Mailbox PeerMessage ->   Peer@@ -178,7 +197,7 @@  -- | Tell the peer process that the node has finished handling one of the -- messages that it published, so that it may read another one.-peerMessageDone :: (MonadIO m) => Peer -> m ()+peerMessageDone :: Peer -> IO () peerMessageDone p = atomically $ modifyTVar' p.pending (max 0 . subtract 1)  -- | Run peer process in current thread. The 'Peer' must be the one that@@ -189,21 +208,22 @@ -- called or the connection is closed, and throws if the peer sends invalid -- data. peer ::-  (MonadUnliftIO m) =>   PeerConfig ->   Peer ->   Inbox PeerMessage ->-  m ()-peer cfg@PeerConfig {label, net, connect, pub, logger} p inbox =-  withRunInIO $ \run -> connect (run . peer_session)+  IO ()+peer cfg@PeerConfig {label, net, pub, logger} p inbox =+  cfg.connect peer_session   where+    -- Commands from the mailbox, as a stream of messages to send.     go = do-      logDebug logger $ label <> " awaiting event..."-      msg <- receive inbox-      dispatchMessage cfg msg >>= bool (return ()) go+      m <- liftIO $ do+        logDebug logger $ label <> " awaiting event..."+        receive inbox+      dispatchMessage cfg m >>= bool (return ()) go     peer_session ad = do-      let ins = transPipe liftIO ad.inboundConduit-          ons = transPipe liftIO ad.outboundConduit+      let ins = ad.inboundConduit+          ons = ad.outboundConduit           src =             runConduit $               ins@@ -214,25 +234,24 @@         link as         runConduit (go .| snk)     -- Wait until the node has room for another message from this peer.-    send_msg msg = do+    send_msg m = do       atomically $ do         n <- readTVar p.pending-        checkSTM (n < maxPendingMessages)+        check (n < maxPendingMessages)         writeTVar p.pending (n + 1)-      publish (PeerMessage p msg) pub+      publish (PeerMessage p m) pub  -- | Internal function to dispatch peer messages. dispatchMessage ::-  (MonadIO m) =>   PeerConfig ->   PeerMessage ->-  ConduitT i Message m Bool-dispatchMessage PeerConfig {label, logger} (SendMessage msg) = do-  logDebug logger (label <> " sending cmd " <> cs (show (msgType msg)))-  yield msg+  ConduitT i Message IO Bool+dispatchMessage PeerConfig {label, logger} (SendMessage m) = do+  liftIO $ logDebug logger (label <> " sending cmd " <> cs (show (msgType m)))+  yield m   return True dispatchMessage PeerConfig {label, logger} KillPeer = do-  logInfo logger (label <> " disconnecting")+  liftIO $ logInfo logger (label <> " disconnecting")   return False  -- | Internal conduit to parse messages coming from peer. The header is checked@@ -245,47 +264,32 @@ -- soon as it reads a count above the limit of the message. The limits are -- 2000 headers and 1000 addresses, for example. inPeerConduit ::-  (MonadIO m) => PeerConfig -> ConduitT ByteString Message m ()+  PeerConfig -> ConduitT ByteString Message IO () inPeerConduit PeerConfig {label, net, logger} = loopForever $ do-  logDebug logger (label <> " awaiting message...")-  x <- takeCE 24 .| foldC-  when (B.null x) $ do-    logError logger (label <> " sent empty message header")-    error "Peer sent empty message header"+  debug (label <> " awaiting message...")+  x <- takeExactly 24   MessageHeader mgc cmd len _ <- case decode x of-    Left _ -> bad "invalid message header"+    Left _ -> bad InvalidMessageHeader     Right h -> return h-  when (mgc /= net.magic) $-    bad ("wrong network magic " <> cs (show mgc))-  when (len > maxPayloadSize net cmd) $-    bad-      ( "too large payload of "-          <> cs (show len)-          <> " bytes for cmd "-          <> cs (show cmd)-      )-  logDebug logger (label <> " sent cmd " <> cs (show cmd))-  y <- takeCE (fromIntegral len) .| foldC-  when (B.length y /= fromIntegral len) $-    bad ("truncated payload for cmd " <> cs (show cmd))+  when (mgc /= net.magic) $ bad (WrongNetwork mgc)+  when (len > maxPayloadSize net cmd) $ bad (MessageTooLarge cmd len)+  debug (label <> " sent cmd " <> cs (show cmd))+  y <- takeExactly (fromIntegral len)   case runGet (getMessage net) $ x `B.append` y of-    Left e ->-      bad-        ( "invalid payload for cmd "-            <> cs (show cmd)-            <> ": "-            <> cs (firstLine e)-        )-    Right msg -> do-      logDebug-        logger-        (label <> " sent payload for cmd " <> cs (show cmd))-      yield msg+    Left e -> bad (InvalidPayload cmd (firstLine e))+    Right m -> do+      debug (label <> " sent payload for cmd " <> cs (show cmd))+      yield m   where-    bad :: (MonadIO m) => Text -> m a-    bad e = do-      logError logger (label <> " sent " <> e)-      error ("Peer sent " <> cs e)+    debug = liftIO . logDebug logger+    takeExactly n = do+      bs <- takeCE n .| foldC+      when (B.length bs /= n) $ bad ConnectionClosed+      return bs+    bad :: PeerException -> ConduitT i o IO a+    bad e = liftIO $ do+      logError logger (label <> ": " <> cs (show e))+      throwIO e     -- The reason is on the first line of the parser's error text. The lines     -- after it are call stack noise.     firstLine = takeWhile (/= '\n')@@ -296,34 +300,56 @@  -- | Ask a peer to disconnect. The peer process stops after sending any messages -- queued before this one.-killPeer :: (MonadIO m) => Peer -> m ()+killPeer :: Peer -> IO () killPeer p = KillPeer `send` p.mailbox  -- | Send a network message to peer. Returns as soon as the message is queued.-sendMessage :: (MonadIO m) => Message -> Peer -> m ()-sendMessage msg p = SendMessage msg `send` p.mailbox---- | Check whether a peer is reserved for a task.-getBusy :: (MonadIO m) => Peer -> m Bool-getBusy p = readTVarIO p.busy+sendMessage :: Message -> Peer -> IO ()+sendMessage m p = SendMessage m `send` p.mailbox --- | Try to reserve a peer for a task, such as syncing headers. Returns 'True' if--- the peer was free and is now reserved, or 'False' if it was already busy.+-- | Try to reserve a peer for a long conversation, such as syncing headers+-- or downloading blocks. Returns 'True' if the peer was free and the owner now+-- holds it, or 'False' if anyone holds it, including the same owner. ----- Reservation exists because some conversations with a peer span many--- messages, like downloading a long run of headers. Two tasks using the same peer--- for such a conversation at once would confuse each other's replies.-setBusy :: (MonadIO m) => Peer -> m Bool-setBusy p =-  atomically $ do-    b <- readTVar p.busy-    unless b $ writeTVar p.busy True-    return $ not b+-- Two tasks holding such a conversation with the same peer at once would+-- confuse each other's replies. "Haskoin.Node.Peer.Lock" has the rules.+reservePeer :: Owner -> Peer -> IO Bool+reservePeer o p = atomically $ do+  lock <- readTVar p.busy+  case acquire o lock of+    Just lock' -> writeTVar p.busy lock' >> return True+    Nothing -> return False --- | Release a peer reserved with 'setBusy'.-setFree :: (MonadIO m) => Peer -> m ()-setFree p = atomically $ writeTVar p.busy False+-- | Release a peer. Returns 'False', and changes nothing, if the owner does not+-- hold it, so that a task cannot free a peer in the middle of another task's+-- conversation.+releasePeer :: Owner -> Peer -> IO Bool+releasePeer o p = atomically $ do+  lock <- readTVar p.busy+  case release o lock of+    Just lock' -> writeTVar p.busy lock' >> return True+    Nothing -> return False +-- | Who holds a peer, if anyone.+peerHolder :: Peer -> IO (Maybe Owner)+peerHolder p = holder <$> readTVarIO p.busy++-- | Check whether a peer is reserved by anyone.+getBusy :: Peer -> IO Bool+getBusy p = isJust <$> peerHolder p++-- | Reserve a peer on behalf of a single owner shared by every caller of this+-- function. Use 'reservePeer' with an owner from 'newOwner' instead.+setBusy :: Peer -> IO Bool+setBusy = reservePeer legacyOwner+{-# DEPRECATED setBusy "Use reservePeer with an owner from newOwner" #-}++-- | Release a peer reserved with 'setBusy'. Peers reserved through+-- 'reservePeer' by another owner are left alone.+setFree :: Peer -> IO ()+setFree = void . releasePeer legacyOwner+{-# DEPRECATED setFree "Use releasePeer with the owner that reserved the peer" #-}+ -- | Request full blocks from peer. Will return 'Nothing' if the list of blocks -- returned by the peer is incomplete, comes out of order, or a timeout is -- reached.@@ -331,17 +357,15 @@ -- The 'Int' argument is the timeout in seconds. On networks with SegWit, -- blocks are requested with their witness data. getBlocks ::-  (MonadUnliftIO m) =>   Network ->   Int ->   Peer ->   [BlockHash] ->-  m (Maybe [Block])+  IO (Maybe [Block]) getBlocks net time p bhs =-  runMaybeT $ mapM f =<< MaybeT (getData time p (GetData ivs))+  (>>= traverse block) <$> getData time p (GetData ivs)   where-    f (Right b) = return b-    f (Left _) = MaybeT $ return Nothing+    block = either (const Nothing) Just     c       | net.segWit = InvWitnessBlock       | otherwise = InvBlock@@ -354,17 +378,15 @@ -- The 'Int' argument is the timeout in seconds. Peers normally only serve -- transactions that are in their mempool. getTxs ::-  (MonadUnliftIO m) =>   Network ->   Int ->   Peer ->   [TxHash] ->-  m (Maybe [Tx])+  IO (Maybe [Tx]) getTxs net time p ths =-  runMaybeT $ mapM f =<< MaybeT (getData time p (GetData ivs))+  (>>= traverse tx) <$> getData time p (GetData ivs)   where-    f (Right _) = MaybeT $ return Nothing-    f (Left t) = return t+    tx = either Just (const Nothing)     c       | net.segWit = InvWitnessTx       | otherwise = InvTx@@ -376,51 +398,27 @@ -- -- The first argument is the timeout in seconds. A ping is sent right after the -- request; since peers answer in order, receiving the matching pong before all--- items means that the peer will not send the rest. Transactions are returned--- as 'Left' and blocks as 'Right', in the order they were requested.+-- items means that the peer will not send the rest. Unrelated messages that+-- arrive meanwhile are ignored. Transactions are returned as 'Left' and blocks+-- as 'Right', in the order they were requested. "Haskoin.Node.Peer.Request"+-- has the rules. getData ::-  (MonadUnliftIO m) => Int -> Peer -> GetData -> m (Maybe [Either Tx Block])+  Int -> Peer -> GetData -> IO (Maybe [Either Tx Block]) getData seconds p gd@(GetData ivs) =   withSubscription p.pub $ \inb -> do     r <- randomIO     MGetData gd `sendMessage` p     MPing (Ping r) `sendMessage` p-    fmap join-      . timeout (seconds * 1000 * 1000)-      . runMaybeT-      $ get_thing inb r [] ivs+    join <$> timeout (seconds * 1000 * 1000) (step inb (newRequest r ivs))   where-    get_thing _inb _r acc [] =-      return $ reverse acc-    get_thing inb r acc hss@(InvVector t h : hs) =-      lift (filterReceive p inb) >>= \case-        MTx tx-          | is_tx t && (txHash tx).get == h ->-              get_thing inb r (Left tx : acc) hs-        MBlock b@(Block bh _)-          | is_block t && (headerHash bh).get == h ->-              get_thing inb r (Right b : acc) hs-        MNotFound (NotFound nvs)-          | not (null (nvs `union` hs)) ->-              MaybeT $ return Nothing-        MPong (Pong r')-          | r == r' ->-              MaybeT $ return Nothing-        _-          | null acc ->-              get_thing inb r acc hss-          | otherwise ->-              MaybeT $ return Nothing-    is_tx InvWitnessTx = True-    is_tx InvTx = True-    is_tx _ = False-    is_block InvWitnessBlock = True-    is_block InvBlock = True-    is_block _ = False+    step inb = \case+      Complete items -> return (Just items)+      Failed _ -> return Nothing+      Continue req -> filterReceive p inb >>= step inb . stepRequest req  -- | Ping a peer and await response. Return 'False' if response not received -- before timeout. The first argument is the timeout in seconds.-pingPeer :: (MonadUnliftIO m) => Int -> Peer -> m Bool+pingPeer :: Int -> Peer -> IO Bool pingPeer time p =   fmap isJust . withSubscription p.pub $ \sub -> do     r <- randomIO@@ -432,8 +430,8 @@  -- | Receive the next message from the given peer, discarding events about other -- peers.-filterReceive :: (MonadIO m) => Peer -> Inbox PeerEvent -> m Message+filterReceive :: Peer -> Inbox PeerEvent -> IO Message filterReceive p inb =   receive inb >>= \case-    PeerMessage p' msg | p == p' -> return msg+    PeerMessage p' m | p == p' -> return m     _ -> filterReceive p inb

@@ -0,0 +1,102 @@+-- |+-- Module      : Haskoin.Node.Peer.Lock+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Exclusive use of a peer, as an explicit state machine.+--+-- Some conversations with a peer span many messages, such as downloading a+-- long run of headers or blocks. Two tasks holding the same conversation with+-- one peer would confuse each other's replies, so a task reserves the peer+-- first. The header chain does this, and so do applications that download+-- blocks, such as Haskoin Store.+--+-- = Why it works this way+--+-- * __Only the holder can release.__ The lock used to be a flag that anyone+--   could clear. A task that released a peer it did not hold, by mistake or+--   after its own reservation had already ended, would free the peer in the+--   middle of another task's conversation. Here a peer is 'Free' or held by an+--   'Owner', and 'release' only succeeds for the holder.+--+-- * __Pure and polymorphic.__ The transitions are plain functions over any+--   owner type with equality, so the rules can be tested with integers.+--   "Haskoin.Node.Peer" keeps a 'Lock' 'Owner' in each 'Haskoin.Node.Peer.Peer'+--   and applies them atomically.+module Haskoin.Node.Peer.Lock+  ( Lock (..),+    acquire,+    release,+    holder,+    Owner,+    newOwner,+    legacyOwner,+    ownerName,+  )+where++import Data.Function (on)+import Data.Text (Text)+import Data.Unique (Unique, newUnique)++-- | Whether a peer is reserved, and by whom.+data Lock o+  = -- | nobody is using the peer for a long conversation+    Free+  | -- | the owner is using the peer+    HeldBy !o+  deriving stock (Eq, Show)++-- | Reserve a free peer. 'Nothing' if it is held, even by the same owner:+-- reserving twice is a mistake that would hide a missing release.+acquire :: o -> Lock o -> Maybe (Lock o)+acquire o = \case+  Free -> Just (HeldBy o)+  HeldBy _ -> Nothing++-- | Release a peer. 'Nothing' unless the owner holds it.+release :: (Eq o) => o -> Lock o -> Maybe (Lock o)+release o = \case+  HeldBy o' | o == o' -> Just Free+  _ -> Nothing++-- | Who holds the peer, if anyone.+holder :: Lock o -> Maybe o+holder = \case+  Free -> Nothing+  HeldBy o -> Just o++-- | A task that reserves peers. Owners are compared by identity, and named+-- for logs.+data Owner+  = Owner !Unique !Text+  | -- | the owner behind the deprecated 'Haskoin.Node.Peer.setBusy' and+    -- 'Haskoin.Node.Peer.setFree'+    Legacy++instance Eq Owner where+  (==) = (==) `on` key+    where+      key (Owner u _) = Just u+      key Legacy = Nothing++instance Show Owner where+  show = show . ownerName++-- | A new owner with a name for logs. Each call gives a different owner.+newOwner :: Text -> IO Owner+newOwner name = (`Owner` name) <$> newUnique++-- | The single owner shared by every caller of the deprecated+-- 'Haskoin.Node.Peer.setBusy' and 'Haskoin.Node.Peer.setFree'.+legacyOwner :: Owner+legacyOwner = Legacy++-- | Name of an owner.+ownerName :: Owner -> Text+ownerName = \case+  Owner _ name -> name+  Legacy -> "legacy"

@@ -0,0 +1,104 @@+-- |+-- Module      : Haskoin.Node.Peer.Request+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- Receiving the answer to a request for blocks or transactions, as an+-- explicit state machine. 'Haskoin.Node.Peer.getData' feeds it the messages+-- that a peer sends after the request.+--+-- = Why it works this way+--+-- Right after a getdata message, the node sends the peer a ping. Peers answer+-- messages in order, so every requested item must arrive, in the order+-- requested, before the matching pong. That gives clear rules:+--+-- * The next requested item moves the request forward.+--+-- * A requested item that is not the next one means the peer skipped+--   something, so the request fails.+--+-- * A notfound message naming a requested item, or the pong arriving early,+--   means the peer will not send everything, so the request fails.+--+-- * Anything else is unrelated traffic and is ignored. Peers send pings,+--   inventory and other chatter whenever they like, and failing a long+--   download because of one of them would only make the application retry it.+--+-- Keeping these rules in a pure function means each one can be tested on its+-- own, without a connection.+module Haskoin.Node.Peer.Request+  ( Request,+    RequestStep (..),+    newRequest,+    stepRequest,+  )+where++import Data.Text (Text)+import Data.Word (Word64)+import Haskoin++-- | A request in progress: the nonce of the ping that follows it, the items+-- still expected, and the items received so far, newest first.+data Request = Request+  { nonce :: !Word64,+    remaining :: ![InvVector],+    received :: ![Either Tx Block]+  }+  deriving stock (Eq, Show)++-- | Outcome of feeding a message to a request.+data RequestStep+  = -- | more items are expected+    Continue !Request+  | -- | every item arrived, in the order requested; transactions are 'Left'+    -- and blocks 'Right'+    Complete ![Either Tx Block]+  | -- | the peer will not send what was requested+    Failed !Text+  deriving stock (Eq, Show)++-- | Start a request for the given items, followed by a ping with the given+-- nonce. A request for nothing is complete at once.+newRequest :: Word64 -> [InvVector] -> RequestStep+newRequest n ivs = check Request {nonce = n, remaining = ivs, received = []}++-- | Feed a message from the peer to a request.+stepRequest :: Request -> Message -> RequestStep+stepRequest r message =+  case (r.remaining, message) of+    ([], _) -> Complete (reverse r.received)+    (next : _, MTx tx)+      | isTx next tx -> accept (Left tx)+      | any (`isTx` tx) r.remaining -> Failed "transaction out of order"+    (next : _, MBlock b)+      | isBlock next b -> accept (Right b)+      | any (`isBlock` b) r.remaining -> Failed "block out of order"+    (_, MNotFound (NotFound ivs))+      | any (`elem` r.remaining) ivs -> Failed "item not found"+    (_, MPong (Pong n))+      | n == r.nonce -> Failed "incomplete response"+    _ -> Continue r+  where+    accept x =+      check r {remaining = drop 1 r.remaining, received = x : r.received}++check :: Request -> RequestStep+check r+  | null r.remaining = Complete (reverse r.received)+  | otherwise = Continue r++-- | Whether a transaction is the one an inventory vector asks for.+isTx :: InvVector -> Tx -> Bool+isTx iv tx =+  iv.invType `elem` [InvTx, InvWitnessTx] && iv.invHash == (txHash tx).get++-- | Whether a block is the one an inventory vector asks for.+isBlock :: InvVector -> Block -> Bool+isBlock iv b =+  iv.invType `elem` [InvBlock, InvWitnessBlock]+    && iv.invHash == (headerHash b.header).get

@@ -1,16 +1,3 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DuplicateRecordFields #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE ImportQualifiedPost #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE MultiWayIf #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TupleSections #-}-{-# LANGUAGE NoFieldSelectors #-}- -- | -- Module      : Haskoin.Node.PeerMgr -- Copyright   : (c) Haskoin Developers@@ -21,8 +8,8 @@ -- -- The peer manager keeps the node connected to the network. ----- It maintains a pool of known peer addresses, filled from the static peers in--- the configuration, from the DNS seeds of the network when discovery is on, and+-- It keeps a book of peer addresses, filled from the static peers in the+-- configuration, from the DNS seeds of the network when discovery is on, and -- from addresses relayed by connected peers. A background loop connects to new -- peers while there are fewer than 'maxPeers' of them. --@@ -31,14 +18,26 @@ -- with a 'PeerConnected' event. Peers are pinged periodically to measure their -- latency, and the list returned by 'getPeers' is sorted by median ping time, so -- that the fastest peers come first. A peer is disconnected if it stops--- responding, does not answer a ping in time, or has been connected for longer--- than 'maxPeerLife'. A 'PeerDisconnected' event is published when an online--- peer goes away.+-- responding, does not answer a ping in time, does not complete the handshake+-- in time, or has been connected for longer than 'maxPeerLife'. A+-- 'PeerDisconnected' event is published when an announced peer goes away. -- -- Most of the @peerMgr*@ functions are used by "Haskoin.Node" to feed messages -- received from peers to the peer manager; applications rarely need to call -- them directly. --+-- = State machines+--+-- The rules live in pure modules, and this one stores their state and carries+-- out what they decide:+--+-- * "Haskoin.Node.PeerMgr.Lifecycle": the life of each connection, from the+--   handshake ("Haskoin.Node.PeerMgr.Handshake") through pings to its+--   disconnection, with a reason.+--+-- * "Haskoin.Node.PeerMgr.Addresses": which address to try next, and how long+--   to wait before trying an address again after its connection ended.+-- -- = Why it works this way -- -- * __Peers are ranked by measured latency.__ The fastest peers deliver@@ -54,6 +53,14 @@ --   longer than 'timeout' is dropped. Waiting for a peer that has gone away --   would stall whatever was waiting on it. --+-- * __A dropped peer is gone at once.__ A peer told to disconnect is no longer+--   offered by 'getPeers' and its messages are no longer routed, even before+--   its connection closes.+--+-- * __Bad addresses wait.__ An address that fails is tried again later each+--   time, and one whose peer broke the protocol is not tried again for a long+--   time, so the node does not hammer dead peers or return to bad ones.+-- -- * __Relayed addresses are untrusted.__ Peers gossip addresses, and a --   malicious peer could gossip private or loopback addresses to make the node --   probe its own network. Only public addresses are accepted, and only a@@ -62,6 +69,9 @@   ( PeerMgrConfig (..),     PeerEvent (..),     OnlinePeer (..),+    PeerState (..),+    Stage (..),+    DisconnectReason (..),     PeerMgr,     withPeerMgr,     peerMgrBest,@@ -83,27 +93,31 @@ where  import Control.Applicative ((<|>))-import Control.Arrow-import Control.Monad-import Data.Bits ((.&.))+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, ExceptionInLinkedThread (..), link, withAsync)+import Control.Concurrent.STM+import Control.Exception (SomeException, fromException)+import Control.Monad (forM_, when, (<=<)) import Data.Function (on)-import Data.List (dropWhileEnd, elemIndex, find, nub, sort)-import Data.Maybe-import Data.Set (Set)-import Data.Set qualified as Set+import Data.List (find, sort)+import Data.Maybe (fromMaybe) import Data.String.Conversions (cs) import Data.Time.Clock import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)-import Data.Word (Word32, Word64, Word8)+import Data.Word (Word32, Word64)+import GHC.Records (HasField (..)) import Haskoin+import Haskoin.Node.Address (isRoutable, toHostService, toSockAddr) import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever) import Haskoin.Node.Peer+import Haskoin.Node.Peer.Lock (Lock (..))+import Haskoin.Node.PeerMgr.Addresses+import Haskoin.Node.PeerMgr.Handshake+import Haskoin.Node.PeerMgr.Lifecycle import NQE-import Network.Socket+import Network.Socket (SockAddr) import System.Random (randomIO, randomRIO)-import UnliftIO-import UnliftIO.Concurrent  -- | Configuration of the peer manager. "Haskoin.Node" builds it from the -- 'Haskoin.Node.NodeConfig'.@@ -143,9 +157,9 @@     mailbox :: !(Mailbox PeerMgrMessage),     -- | height of the best block, sent to peers in version messages     best :: !(TVar BlockHeight),-    -- | known addresses not yet connected to, at most 'maxKnownAddresses'-    addresses :: !(TVar (Set SockAddr)),-    -- | peers being connected to or online, fastest first+    -- | addresses to connect to+    addresses :: !(TVar (AddressBook SockAddr)),+    -- | open connections, fastest peers first     peers :: !(TVar [OnlinePeer])   } @@ -153,10 +167,10 @@ data PeerMgrMessage   = -- | connect to a peer address     Connect !SockAddr-  | -- | check whether a peer timed out or must be pinged+  | -- | check the timers of a peer     CheckPeer !Peer-  | -- | a peer process stopped-    PeerDied !Child+  | -- | a peer process stopped, with the exception that stopped it+    PeerDied !Child !(Maybe SomeException)   | -- | the height of the best block changed     ManagerBest !BlockHeight   | -- | a peer acknowledged our version@@ -172,181 +186,177 @@   | -- | reply once every previous message has been handled     Sync !(Listen ()) --- | Data structure representing an online peer.+-- | A connection to a peer. -- -- Peers are compared for equality by their 'Peer' handle, and ordered by the -- median of their recent ping times, so that sorting a list puts the fastest -- peers first. Peers without any ping measurement sort as if their median ping -- were 60 seconds.+--+-- The fields @version@, @online@, @verack@, @ping@, @pings@, @connected@ and+-- @tickled@ of earlier releases are now computed from 'state', so record dot+-- syntax such as @o.version@ keeps working. They cannot be set or pattern+-- matched. data OnlinePeer = OnlinePeer   { -- | network address of the peer     address :: !SockAddr,-    -- | whether the peer acknowledged our version-    verack :: !Bool,-    -- | whether the handshake is complete: both version and verack received-    online :: !Bool,-    -- | version message sent by the peer, if received-    version :: !(Maybe Version),+    -- | stage of the connection and its timers; see+    -- "Haskoin.Node.PeerMgr.Lifecycle"+    state :: !PeerState,     -- | thread running the peer process     async :: !(Async ()),     -- | handle to talk to the peer     mailbox :: !Peer,     -- | nonce of the version message we sent, used to detect connections to     -- ourselves-    nonce :: !Word64,-    -- | time and nonce of the ping awaiting an answer, if any-    ping :: !(Maybe (UTCTime, Word64)),-    -- | round-trip times of the last eleven pings, sorted-    pings :: ![NominalDiffTime],-    -- | when the connection was made-    connected :: !UTCTime,-    -- | when the peer last sent a message-    tickled :: !UTCTime+    nonce :: !Word64   }  instance Eq OnlinePeer where-  (==) = (==) `on` f-    where-      f OnlinePeer {mailbox = p} = p+  (==) = (==) `on` (.mailbox)  instance Ord OnlinePeer where-  compare = compare `on` f-    where-      f OnlinePeer {pings = pings} = fromMaybe 60 (median pings)+  compare = compare `on` \o -> fromMaybe 60 (median o.state.pings) +-- | Version message sent by the peer, if received.+instance HasField "version" OnlinePeer (Maybe Version) where+  getField o = stageVersion o.state++-- | Whether the peer is online: the handshake is complete and the peer is not+-- being disconnected.+instance HasField "online" OnlinePeer Bool where+  getField o = isOnline o.state++-- | Whether the peer acknowledged our version.+instance HasField "verack" OnlinePeer Bool where+  getField o = case o.state.stage of+    Handshaking h -> hasVerAck h+    Online _ _ -> True+    Disconnecting _ -> False++-- | Time and nonce of the ping awaiting an answer, if any.+instance HasField "ping" OnlinePeer (Maybe (UTCTime, Word64)) where+  getField o = pendingPing o.state++-- | Round-trip times of the last eleven pings, sorted.+instance HasField "pings" OnlinePeer [NominalDiffTime] where+  getField o = sort o.state.pings++-- | When the connection was opened.+instance HasField "connected" OnlinePeer UTCTime where+  getField o = o.state.connected++-- | When the peer last sent a message.+instance HasField "tickled" OnlinePeer UTCTime where+  getField o = o.state.lastHeard+ -- | Start a peer manager and pass it to the given action. The peer manager and -- all its peers are stopped when the action returns. -- -- The peer manager does not connect to any peer until it learns the height of -- the best block through 'peerMgrBest'. "Haskoin.Node" does this as soon as the -- chain is loaded.-withPeerMgr ::-  (MonadUnliftIO m) =>-  PeerMgrConfig ->-  (PeerMgr -> m a) ->-  m a+withPeerMgr :: PeerMgrConfig -> (PeerMgr -> IO a) -> IO a withPeerMgr cfg action = do-  ibx <- newInbox-  withSupervisor (Notify (death ibx)) $ \sup -> do-    bb <- newTVarIO 0-    kp <- newTVarIO Set.empty-    ob <- newTVarIO []+  inbox <- newInbox+  let died (a, e) = PeerDied a e `sendSTM` inbox+  withSupervisor (Notify died) $ \sup -> do+    best <- newTVarIO 0+    addresses <- newTVarIO emptyBook+    peers <- newTVarIO []     let mgr =           PeerMgr             { config = cfg,               supervisor = sup,-              mailbox = inboxToMailbox ibx,-              best = bb,-              addresses = kp,-              peers = ob+              mailbox = inboxToMailbox inbox,+              best = best,+              addresses = addresses,+              peers = peers             }-    go mgr ibx-  where-    death ibx (a, _e) = PeerDied a `sendSTM` ibx-    go mgr ibx = withAsync (peerManager mgr ibx) $ \a ->+    withAsync (peerManager mgr inbox) $ \a ->       withConnectLoop mgr (link a >> action mgr)  -- | Main loop of the peer manager process. Waits for the best block height, then -- handles messages forever.-peerManager ::-  (MonadUnliftIO m) => PeerMgr -> Inbox PeerMgrMessage -> m ()-peerManager mgr ibx = do+peerManager :: PeerMgr -> Inbox PeerMgrMessage -> IO ()+peerManager mgr inbox = do   logDebug mgr.config.logger "Getting best block..."-  putBestBlock mgr <=< receiveMatch ibx $ \case+  (atomically . writeTVar mgr.best) <=< receiveMatch inbox $ \case     ManagerBest b -> Just b     _ -> Nothing   loopForever $ do     logDebug mgr.config.logger "Awaiting event..."-    dispatch mgr =<< receive ibx---- | Store the height of the best block.-putBestBlock :: (MonadIO m) => PeerMgr -> BlockHeight -> m ()-putBestBlock mgr bb = atomically $ writeTVar mgr.best bb---- | Read the height of the best block.-getBestBlock :: (MonadIO m) => PeerMgr -> m BlockHeight-getBestBlock mgr = readTVarIO mgr.best---- | Network constants of the peer manager.-getNetwork :: PeerMgr -> Network-getNetwork mgr = mgr.config.net---- | Add the static peers and, if discovery is on, the network seeds to the known addresses.-loadPeers :: (MonadIO m) => PeerMgr -> m ()-loadPeers mgr = do-  loadStaticPeers mgr-  loadNetSeeds mgr---- | Resolve the static peers in the configuration and add them to the known addresses.-loadStaticPeers :: (MonadIO m) => PeerMgr -> m ()-loadStaticPeers mgr =-  mapM_ (newPeer mgr) . concat-    =<< mapM (liftIO . toSockAddr mgr.config.net) mgr.config.peers---- | If discovery is on, resolve the DNS seeds of the network and add them to the known addresses.-loadNetSeeds :: (MonadIO m) => PeerMgr -> m ()-loadNetSeeds mgr =-  when mgr.config.discover $ do-    ss <--      concat-        <$> mapM (liftIO . toSockAddr mgr.config.net) mgr.config.net.seeds-    mapM_ (newPeer mgr) ss+    dispatch mgr =<< receive inbox --- | Log how many peers are online out of the maximum.-logConnectedPeers :: (MonadIO m) => PeerMgr -> m ()-logConnectedPeers mgr = do-  let m = mgr.config.maxPeers-  l <- length <$> getConnectedPeers mgr-  logInfo-    mgr.config.logger-    ("Peers connected: " <> cs (show l) <> "/" <> cs (show m))+-- | Time limits of connections.+limits :: PeerMgr -> Limits+limits mgr = Limits {timeout = mgr.config.timeout, maxLife = mgr.config.maxPeerLife} --- | All peers being connected to or online, fastest first.-getOnlinePeers :: (MonadIO m) => PeerMgr -> m [OnlinePeer]-getOnlinePeers mgr = readTVarIO mgr.peers+-- | Handle a message sent to the peer manager.+dispatch :: PeerMgr -> PeerMgrMessage -> IO ()+dispatch mgr = \case+  PeerVersion p v -> do+    logDebug mgr.config.logger ("Received version from peer " <> p.label)+    nonces <- map (.nonce) <$> readTVarIO mgr.peers+    peerInput mgr p $ either Refused (GotHandshake . GotVersion) (checkVersion nonces v)+  PeerVerAck p -> do+    logDebug mgr.config.logger ("Received version ack from peer " <> p.label)+    peerInput mgr p (GotHandshake GotVerAck)+  PeerPong p n -> do+    now <- getCurrentTime+    peerInput mgr p (GotPong now n)+  CheckPeer p -> do+    now <- getCurrentTime+    nonce <- randomIO+    peerInput mgr p (Tick now nonce)+  PeerPing p n -> MPong (Pong n) `sendMessage` p+  PeerAddrs p nas -> relayedAddresses mgr p nas+  ManagerBest h -> atomically $ writeTVar mgr.best h+  Connect sa -> connectPeer mgr sa+  PeerDied a e -> peerDied mgr a e+  Sync r -> atomically (r ()) --- | Peers that completed the handshake, fastest first.-getConnectedPeers :: (MonadIO m) => PeerMgr -> m [OnlinePeer]-getConnectedPeers mgr = filter (.online) <$> getOnlinePeers mgr+-- | Apply an input to the state of a connection and carry out the resulting+-- actions. Inputs for connections that are gone are ignored.+peerInput :: PeerMgr -> Peer -> PeerInput -> IO ()+peerInput mgr p input = do+  result <- atomically $ do+    ops <- readTVar mgr.peers+    case find ((== p) . (.mailbox)) ops of+      Nothing -> return Nothing+      Just o -> do+        let (st, actions) = stepPeer (limits mgr) input o.state+            o' = o {state = st}+        writeTVar mgr.peers (sort (o' : filter (/= o) ops))+        return (Just (o', actions))+  forM_ result $ \(o, actions) -> forM_ actions (perform o)+  where+    perform o = \case+      ReplyVerAck -> MVerAck `sendMessage` p+      Announce -> do+        atomically $ modifyTVar' mgr.addresses (markConnected o.address)+        logInfo mgr.config.logger ("Connected to peer " <> p.label)+        publish (PeerConnected p) mgr.config.pub+        logConnectedPeers mgr+      SendPing n -> MPing (Ping n) `sendMessage` p+      Disconnect r -> do+        logWarn+          mgr.config.logger+          ("Disconnecting peer " <> p.label <> ": " <> cs (show r))+        killPeer p --- | Publish a peer event.-managerEvent :: (MonadIO m) => PeerMgr -> PeerEvent -> m ()-managerEvent mgr e = publish e mgr.config.pub+-- | Record that a peer just sent a message, so that it is not disconnected for+-- being idle. "Haskoin.Node" calls this for every message received.+ticklePeer :: PeerMgr -> Peer -> IO ()+ticklePeer mgr p = getCurrentTime >>= peerInput mgr p . Heard --- | Handle a message sent to the peer manager.-dispatch ::-  (MonadUnliftIO m) => PeerMgr -> PeerMgrMessage -> m ()-dispatch mgr (PeerVersion p v) = do-  logDebug-    mgr.config.logger-    ("Received peer " <> p.label <> " version " <> cs (show v))-  let b = mgr.peers-  atomically (setPeerVersion b p v) >>= \case-    Just o -> do-      when o.online (announcePeer mgr p)-      logDebug-        mgr.config.logger-        ("Sending version ack to peer " <> p.label)-      MVerAck `sendMessage` p-    Nothing -> do-      logWarn-        mgr.config.logger-        ("Version rejected for peer " <> p.label)-      killPeer p-dispatch mgr (PeerVerAck p) = do-  atomically (setPeerVerAck mgr.peers p) >>= \case-    Just o -> do-      logDebug mgr.config.logger ("Received version ack from peer " <> p.label)-      when o.online (announcePeer mgr p)-    Nothing -> do-      logWarn-        mgr.config.logger-        ("Verack rejected for peer " <> p.label)-      killPeer p-dispatch mgr (PeerAddrs p nas)+-- | Learn the routable addresses that an online peer relayed, if discovery is+-- on.+relayedAddresses :: PeerMgr -> Peer -> [NetworkAddress] -> IO ()+relayedAddresses mgr p nas   | mgr.config.discover =       getOnlinePeer mgr p >>= \case-        Just OnlinePeer {online = True} -> do+        Just o | isOnline o.state -> do           let sas = filter isRoutable $ map (hostToSockAddr . (.address)) nas           logDebug             mgr.config.logger@@ -358,451 +368,212 @@                 <> cs (show (length sas))                 <> " routable"             )-          forM_ sas (newPeer mgr)+          atomically . modifyTVar' mgr.addresses $ \book ->+            foldr (addAddress Discovered) book sas         _ ->           logDebug             mgr.config.logger-            ("Ignoring addresses from peer before handshake " <> p.label)+            ("Ignoring addresses from peer that is not online " <> p.label)   | otherwise =       logDebug mgr.config.logger ("Ignoring addresses from peer " <> p.label)-dispatch _ (Sync r) =-  atomically (r ())-dispatch mgr (PeerPong p n) = do-  logDebug-    mgr.config.logger-    ("Received pong " <> cs (show n) <> " from peer " <> p.label)-  now <- liftIO getCurrentTime-  atomically (gotPong mgr.peers n now p)-dispatch mgr (PeerPing p n) = do-  logDebug-    mgr.config.logger-    ("Responding to ping " <> cs (show n) <> " from peer " <> p.label)-  MPong (Pong n) `sendMessage` p-dispatch mgr (ManagerBest h) = do-  logDebug mgr.config.logger ("Setting best block to " <> cs (show h))-  putBestBlock mgr h-dispatch mgr (Connect sa) = do-  connectPeer mgr sa-dispatch mgr (PeerDied a) = do-  processPeerOffline mgr a-dispatch mgr (CheckPeer p) = do-  logDebug mgr.config.logger ("Housekeeping for peer " <> p.label)-  checkPeer mgr p --- | Record that a peer just sent a message, so that it is not disconnected for--- being idle. "Haskoin.Node" calls this for every message received.-ticklePeer :: (MonadIO m) => PeerMgr -> Peer -> m ()-ticklePeer mgr p = do-  logDebug mgr.config.logger ("Tickling peer " <> p.label)-  t <- liftIO getCurrentTime-  atomically (modifyPeer mgr.peers p (\o -> o {tickled = t}))---- | Disconnect a peer that has been connected for too long, has been idle for--- too long, or did not answer a ping in time. Otherwise ping it if no ping is--- pending.-checkPeer :: (MonadIO m) => PeerMgr -> Peer -> m ()-checkPeer mgr p =-  atomically (mgr.peers `findPeer` p) >>= \case-    Just o -> do-      now <- liftIO getCurrentTime-      let expired = mgr.config.maxPeerLife `addUTCTime` o.connected-          deadline = mgr.config.timeout `addUTCTime` o.tickled-          -- Other traffic must not keep a peer that ignores our ping.-          pingDeadline = (\(t, _) -> mgr.config.timeout `addUTCTime` t) <$> o.ping-      if-        | now > expired -> do-            logError mgr.config.logger ("Peer " <> p.label <> " is too old")-            killPeer p-        | now > deadline -> do-            logError mgr.config.logger ("Peer " <> p.label <> " timed out")-            killPeer p-        | maybe False (now >) pingDeadline -> do-            logError mgr.config.logger ("Peer " <> p.label <> " did not answer ping")-            killPeer p-        | isNothing o.ping -> sendPing mgr p-        | otherwise -> return ()-    Nothing -> do-      logError mgr.config.logger ("Peer " <> p.label <> " is offline")+-- | Forget a connection whose process stopped, decide when to try its address+-- again, and publish its disconnection if it had been announced.+peerDied :: PeerMgr -> Child -> Maybe SomeException -> IO ()+peerDied mgr a e = do+  now <- getCurrentTime+  gone <- atomically $ do+    ops <- readTVar mgr.peers+    case find ((== a) . (.async)) ops of+      Nothing -> return Nothing+      Just o -> do+        let outcome = connectionOutcome o.state e+        writeTVar mgr.peers (filter (/= o) ops)+        modifyTVar' mgr.addresses (finishAddress now outcome o.address)+        return (Just (o, outcome))+  case gone of+    Nothing -> logError mgr.config.logger "Unknown peer process stopped"+    Just (o, outcome) -> do+      logInfo mgr.config.logger $+        "Connection to peer "+          <> o.mailbox.label+          <> " ended: "+          <> cs (show outcome)+          <> maybe "" (\x -> " (" <> cs (show x) <> ")") e+      when o.state.announced $+        publish (PeerDisconnected o.mailbox) mgr.config.pub+      logConnectedPeers mgr --- | Send a ping with a random nonce to an online peer and remember when it was sent.-sendPing :: (MonadIO m) => PeerMgr -> Peer -> m ()-sendPing mgr p = do-  atomically (mgr.peers `findPeer` p) >>= \case-    Nothing ->-      logWarn mgr.config.logger ("Will not ping unknown peer " <> p.label)-    Just o-      | o.online -> do-          n <- randomIO-          now <- liftIO getCurrentTime-          atomically (setPeerPing mgr.peers n now p)-          logDebug-            mgr.config.logger-            ("Sending ping " <> cs (show n) <> " to peer " <> p.label)-          MPing (Ping n) `sendMessage` p-      | otherwise ->-          logDebug-            mgr.config.logger-            ("Will not ping offline peer " <> p.label)+-- | How a connection ended, from its last state and the exception that stopped+-- its process, if any.+--+-- A peer process only stops without an exception when told to disconnect. If+-- the peer manager did not do that, another process did, such as the chain+-- after invalid headers, so the address waits as after a failure.+connectionOutcome :: PeerState -> Maybe SomeException -> Outcome+connectionOutcome st e =+  case (st.stage, peerException =<< e) of+    (Disconnecting TooOld, _) -> Disconnected+    (Disconnecting (Rejected _), _) -> Misbehaved+    (Disconnecting _, _) -> Failed+    (_, Just ConnectionClosed)+      | st.announced -> Disconnected+      | otherwise -> Failed+    (_, Just _) -> Misbehaved+    (_, Nothing) -> Failed+  where+    peerException x =+      fromException x+        <|> (fromException x >>= \(ExceptionInLinkedThread _ inner) -> fromException inner) --- | Clean up after a peer process stops, and publish a disconnection event if the peer was online.-processPeerOffline :: (MonadIO m) => PeerMgr -> Child -> m ()-processPeerOffline mgr a = do-  atomically (findPeerAsync mgr.peers a) >>= \case-    Nothing -> logError mgr.config.logger "Disconnected unknown peer"-    Just o -> do-      if o.online-        then do-          logError-            mgr.config.logger-            ("Disconnected peer " <> o.mailbox.label)-          managerEvent mgr (PeerDisconnected o.mailbox)-        else-          logError-            mgr.config.logger-            ("Could not connect to peer " <> o.mailbox.label)-      atomically (removePeer mgr.peers o.mailbox)-      logConnectedPeers mgr+-- | Log how many peers are online out of the maximum.+logConnectedPeers :: PeerMgr -> IO ()+logConnectedPeers mgr = do+  l <- length <$> getPeers mgr+  logInfo+    mgr.config.logger+    ("Peers connected: " <> cs (show l) <> "/" <> cs (show mgr.config.maxPeers)) --- | Publish a connection event for a peer that just completed the handshake.-announcePeer :: (MonadIO m) => PeerMgr -> Peer -> m ()-announcePeer mgr p = do-  atomically (findPeer mgr.peers p) >>= \case-    Just OnlinePeer {online = True} -> do-      logInfo mgr.config.logger ("Connected to peer " <> p.label)-      managerEvent mgr (PeerConnected p)-      logConnectedPeers mgr-    Just OnlinePeer {online = False} ->-      return ()-    Nothing ->-      logWarn mgr.config.logger ("Not announcing disconnected peer " <> p.label)+-- | Next address to connect to. The configured peers and, if discovery is on,+-- the seeds are loaded when there is nothing to try and they were not loaded+-- recently.+getNewPeer :: PeerMgr -> IO (Maybe SockAddr)+getNewPeer mgr = do+  now <- getCurrentTime+  reload <- needsReload now <$> readTVarIO mgr.addresses+  when reload $ loadPeers mgr now+  atomically . stateTVar mgr.addresses $ \book ->+    case takeAddress now book of+      Just (sa, book') -> (Just sa, book')+      Nothing -> (Nothing, book) --- | Take an address from the known addresses. If there are none, reload the static peers and seeds first.-getNewPeer :: (MonadIO m) => PeerMgr -> m (Maybe SockAddr)-getNewPeer mgr =-  g >>= \case-    Nothing -> loadPeers mgr >> g-    Just a -> return (Just a)-  where-    g =-      atomically $-        stateTVar-          mgr.addresses-          (first (listToMaybe . Set.elems) . Set.splitAt 1)+-- | Resolve the configured peers and, if discovery is on, the seeds of the+-- network, and add them to the address book.+loadPeers :: PeerMgr -> UTCTime -> IO ()+loadPeers mgr now = do+  let net = mgr.config.net+      resolve = fmap concat . mapM (toSockAddr net)+  configured <- resolve mgr.config.peers+  seeds <- if mgr.config.discover then resolve net.seeds else return []+  atomically . modifyTVar' mgr.addresses $ \book ->+    markLoaded now $+      foldr (addAddress Discovered) (foldr (addAddress Configured) book configured) seeds --- | Start a peer process for an address and send it our version, unless there--- are no free slots or the address is already connected.-connectPeer :: (MonadUnliftIO m) => PeerMgr -> SockAddr -> m ()+-- | Start a peer process for an address taken from the address book and send+-- it our version, unless there are no free slots or the address is connected+-- already, in which case the address is given back.+connectPeer :: PeerMgr -> SockAddr -> IO () connectPeer mgr sa = do-  n <- length <$> getOnlinePeers mgr-  atomically (findPeerAddress mgr.peers sa) >>= \case-    _-      | n >= mgr.config.maxPeers ->-          -- The connect loop may have queued this before the last free slot-          -- was taken.-          logDebug-            mgr.config.logger-            ("Not connecting to peer " <> cs (show sa) <> ": no free slots")-    Just _ ->-      logWarn mgr.config.logger ("Duplicate connection to peer " <> cs (show sa))-    Nothing -> do-      logInfo mgr.config.logger ("Connecting to peer " <> cs (show sa))-      nonce <- randomIO-      bb <- getBestBlock mgr-      now <- liftIO getCurrentTime-      let rmt = NetworkAddress (srv mgr.config.net) (sockToHostAddress sa)-          unix = floor (utcTimeToPOSIXSeconds now)-          ver = buildVersion mgr.config.net nonce bb mgr.config.address rmt unix-          text = cs (show sa)-      (inbox, mailbox) <- newMailbox-      let pc =-            PeerConfig-              { pub = mgr.config.pub,-                net = mgr.config.net,-                label = text,-                connect = mgr.config.connect sa,-                logger = withSource "Peer" mgr.config.logger+  ops <- readTVarIO mgr.peers+  if+    | length ops >= mgr.config.maxPeers -> giveBack "no free slots"+    | any ((== sa) . (.address)) ops -> giveBack "already connected"+    | otherwise -> do+        logInfo mgr.config.logger ("Connecting to peer " <> cs (show sa))+        nonce <- randomIO+        height <- readTVarIO mgr.best+        now <- getCurrentTime+        let net = mgr.config.net+            services = if net.segWit then 8 else 0+            rmt = NetworkAddress services (sockToHostAddress sa)+            unix = floor (utcTimeToPOSIXSeconds now)+            ver = buildVersion net nonce height mgr.config.address rmt unix+            pc =+              PeerConfig+                { pub = mgr.config.pub,+                  net = net,+                  label = cs (show sa),+                  connect = mgr.config.connect sa,+                  logger = withSource "Peer" mgr.config.logger+                }+        (inbox, mailbox) <- newMailbox+        busy <- newTVarIO Free+        pending <- newTVarIO 0+        let p = wrapPeer pc busy pending mailbox+        -- The peer manager handles messages one at a time, so the peer cannot+        -- die before it is in the list.+        a <- mgr.supervisor `addChild` withHousekeeping mgr p (peer pc p inbox)+        MVersion ver `sendMessage` p+        atomically . modifyTVar' mgr.peers $ \x ->+          sort $+            OnlinePeer+              { address = sa,+                state = newPeerState now,+                async = a,+                mailbox = p,+                nonce = nonce               }-      busy <- newTVarIO False-      pending <- newTVarIO 0-      let p = wrapPeer pc busy pending mailbox-      a <- withRunInIO $ \io ->-        mgr.supervisor `addChild` io (launch pc inbox p)-      MVersion ver `sendMessage` p-      atomically $-        insertPeer-          mgr.peers-          OnlinePeer-            { address = sa,-              verack = False,-              online = False,-              version = Nothing,-              async = a,-              mailbox = p,-              nonce = nonce,-              pings = [],-              ping = Nothing,-              connected = now,-              tickled = now-            }+              : x   where-    srv net-      | net.segWit = 8-      | otherwise = 0-    launch pc inbox p =-      withPeerLoop mgr p (\a -> link a >> peer pc p inbox)+    giveBack why = do+      logDebug+        mgr.config.logger+        ("Not connecting to peer " <> cs (show sa) <> ": " <> why)+      atomically $ modifyTVar' mgr.addresses (returnAddress sa) --- | Run a loop in the background that asks the peer manager to check a peer at--- random intervals between a quarter and a half of the timeout.-withPeerLoop :: (MonadUnliftIO m) => PeerMgr -> Peer -> (Async a -> m a) -> m a-withPeerLoop mgr p =-  withAsync . loopForever $ do-    let timeout = mgr.config.timeout-        ms = floor (timeout * 1000 * 1000)-    r <- randomRIO (ms `div` 4, ms `div` 2)-    threadDelay r-    managerCheck mgr p+-- | Run a peer process along with a loop that asks the peer manager to check+-- the peer's timers at random intervals between a quarter and a half of the+-- timeout.+withHousekeeping :: PeerMgr -> Peer -> IO () -> IO ()+withHousekeeping mgr p go =+  withAsync checks $ \a -> link a >> go+  where+    micros = floor (mgr.config.timeout * 1000000) :: Int+    checks = loopForever $ do+      threadDelay =<< randomRIO (micros `div` 4, micros `div` 2)+      CheckPeer p `send` mgr.mailbox  -- | Run a loop in the background that connects to a new peer every 0.1 to 5 -- seconds while there are free slots.-withConnectLoop :: (MonadUnliftIO m) => PeerMgr -> m a -> m a+withConnectLoop :: PeerMgr -> IO a -> IO a withConnectLoop mgr act =   withAsync go $ \a -> link a >> act   where     go = loopForever $ do-      l <- length <$> getOnlinePeers mgr-      when-        (l < mgr.config.maxPeers)-        (getNewPeer mgr >>= mapM_ (managerConnect mgr))-      threadDelay =<< randomRIO (10 ^ 5, 5 * 10 ^ 6)---- | Upper bound on addresses remembered for future connections. Peers keep--- relaying addresses for as long as the node runs, and with every connection--- slot in use nothing ever removes them, so the pool must be bounded.-maxKnownAddresses :: Int-maxKnownAddresses = 4096---- | Add an address to the known addresses, unless it is connected already or the pool is full.-newPeer :: (MonadIO m) => PeerMgr -> SockAddr -> m ()-newPeer mgr sa =-  atomically $-    findPeerAddress mgr.peers sa >>= \case-      Just _ -> return ()-      Nothing -> modifyTVar' mgr.addresses $ \known ->-        if Set.size known >= maxKnownAddresses-          then known-          else Set.insert sa known---- | Record the round-trip time of a ping if the nonce matches the pending one.-gotPong :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM ()-gotPong b nonce now p =-  findPeer b p >>= \case-    Just o@OnlinePeer {ping = Just (time, nonce')} | nonce' == nonce -> do-      let d = now `diffUTCTime` time-          o' = o {ping = Nothing, pings = sort (take 11 (d : o.pings))}-      insertPeer b o'-    _ -> return ()---- | Remember the time and nonce of a ping sent to a peer.-setPeerPing :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM ()-setPeerPing b nonce now p =-  modifyPeer b p $ \o -> o {ping = Just (now, nonce)}---- | Store the version of a peer. Returns 'Nothing' to reject it if the peer does--- not serve the full chain, is ourselves, is unknown, or already sent a version.------ A peer must offer the full chain ('nodeNetwork') to be useful for syncing, and--- a version whose nonce matches one of ours means the node has connected to--- itself, which happens when it learns its own address from gossip.-setPeerVersion :: TVar [OnlinePeer] -> Peer -> Version -> STM (Maybe OnlinePeer)-setPeerVersion b p v-  | v.services .&. nodeNetwork == 0 = return Nothing-  | otherwise =-      readTVar b >>= \ops ->-        if any (\o -> v.nonce == o.nonce) ops-          then return Nothing -- peer is myself-          else-            findPeer b p >>= \case-              Nothing -> return Nothing -- peer not found-              Just OnlinePeer {version = Just _} ->-                return Nothing -- duplicate version-              Just o -> do-                let n = o {version = Just v, online = o.verack}-                insertPeer b n-                return (Just n)---- | Record that a peer acknowledged our version. Returns 'Nothing' if the peer is--- unknown or already sent one.-setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)-setPeerVerAck b p =-  findPeer b p >>= \case-    Just OnlinePeer {verack = True} -> return Nothing -- duplicate verack-    Just o -> do-      let o' = o {verack = True, online = isJust o.version}-      insertPeer b o'-      return (Just o')-    Nothing -> return Nothing---- | Find a peer in the list.-findPeer :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)-findPeer b p =-  find ((== p) . (.mailbox))-    <$> readTVar b---- | Add or replace a peer in the list, keeping it sorted.-insertPeer :: TVar [OnlinePeer] -> OnlinePeer -> STM ()-insertPeer b o =-  modifyTVar b $ \x -> sort . nub $ o : x---- | Change a peer in the list, if it is there.-modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM ()-modifyPeer b p f =-  findPeer b p >>= \case-    Nothing -> return ()-    Just o -> insertPeer b $ f o---- | Remove a peer from the list.-removePeer :: TVar [OnlinePeer] -> Peer -> STM ()-removePeer b p = modifyTVar b (filter ((/= p) . (.mailbox)))---- | Find a peer by the thread that runs it.-findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer)-findPeerAsync b a = find ((== a) . (.async)) <$> readTVar b---- | Find a peer by its address.-findPeerAddress :: TVar [OnlinePeer] -> SockAddr -> STM (Maybe OnlinePeer)-findPeerAddress b a = find ((== a) . (.address)) <$> readTVar b---- | Peers that completed the handshake, sorted so that those with the lowest--- median ping come first.-getPeers :: (MonadIO m) => PeerMgr -> m [OnlinePeer]-getPeers = getConnectedPeers---- | Information about a peer, or 'Nothing' if the peer manager does not know it.--- The peer may still be in the middle of the handshake; check 'online'.-getOnlinePeer :: (MonadIO m) => PeerMgr -> Peer -> m (Maybe OnlinePeer)-getOnlinePeer mgr p = atomically (mgr.peers `findPeer` p)+      n <- length <$> readTVarIO mgr.peers+      when (n < mgr.config.maxPeers) $+        getNewPeer mgr >>= mapM_ (\sa -> Connect sa `send` mgr.mailbox)+      threadDelay =<< randomRIO (100000 :: Int, 5000000) --- | Ask the peer manager to check a peer.-managerCheck :: (MonadIO m) => PeerMgr -> Peer -> m ()-managerCheck mgr p = CheckPeer p `send` mgr.mailbox+-- | Peers that are online, sorted so that those with the lowest median ping+-- come first. Peers in the handshake or being disconnected are left out.+getPeers :: PeerMgr -> IO [OnlinePeer]+getPeers mgr = filter (isOnline . (.state)) <$> readTVarIO mgr.peers --- | Ask the peer manager to connect to an address.-managerConnect :: (MonadIO m) => PeerMgr -> SockAddr -> m ()-managerConnect mgr sa = Connect sa `send` mgr.mailbox+-- | Information about a connection, or 'Nothing' if the peer manager does not+-- know it. The peer may still be in the handshake, or being disconnected;+-- check its 'state'.+getOnlinePeer :: PeerMgr -> Peer -> IO (Maybe OnlinePeer)+getOnlinePeer mgr p = find ((== p) . (.mailbox)) <$> readTVarIO mgr.peers  -- | Tell the peer manager the height of the best block.-peerMgrBest :: (MonadIO m) => PeerMgr -> BlockHeight -> m ()+peerMgrBest :: PeerMgr -> BlockHeight -> IO () peerMgrBest mgr bh = ManagerBest bh `send` mgr.mailbox  -- | Tell the peer manager that a peer acknowledged our version.-peerMgrVerAck :: (MonadIO m) => PeerMgr -> Peer -> m ()+peerMgrVerAck :: PeerMgr -> Peer -> IO () peerMgrVerAck mgr p = PeerVerAck p `send` mgr.mailbox  -- | Tell the peer manager that a peer sent its version.-peerMgrVersion :: (MonadIO m) => PeerMgr -> Peer -> Version -> m ()+peerMgrVersion :: PeerMgr -> Peer -> Version -> IO () peerMgrVersion mgr p ver = PeerVersion p ver `send` mgr.mailbox  -- | Tell the peer manager that a peer sent a ping, so that it answers with a pong.-peerMgrPing :: (MonadIO m) => PeerMgr -> Peer -> Word64 -> m ()+peerMgrPing :: PeerMgr -> Peer -> Word64 -> IO () peerMgrPing mgr p nonce = PeerPing p nonce `send` mgr.mailbox  -- | Tell the peer manager that a peer answered a ping.-peerMgrPong :: (MonadIO m) => PeerMgr -> Peer -> Word64 -> m ()+peerMgrPong :: PeerMgr -> Peer -> Word64 -> IO () peerMgrPong mgr p nonce = PeerPong p nonce `send` mgr.mailbox  -- | Tell the peer manager about addresses relayed by a peer. They are added to -- the known addresses if discovery is on and they are routable.-peerMgrAddrs :: (MonadIO m) => PeerMgr -> Peer -> [NetworkAddress] -> m ()+peerMgrAddrs :: PeerMgr -> Peer -> [NetworkAddress] -> IO () peerMgrAddrs mgr p addrs = PeerAddrs p addrs `send` mgr.mailbox  -- | Wait until the peer manager has handled every message sent to it before.-peerMgrSync :: (MonadIO m) => PeerMgr -> m ()+peerMgrSync :: PeerMgr -> IO () peerMgrSync mgr = Sync `query` mgr.mailbox---- | Whether an address that a peer relayed is worth connecting to: a public--- unicast address with a port. Loopback, private, link-local, shared,--- documentation, multicast and reserved ranges are refused, so that a peer--- cannot make the node connect to services on its own host or network.--- Static peers in the configuration are not checked.-isRoutable :: SockAddr -> Bool-isRoutable (SockAddrInet port h) =-  port /= 0 && routable4 (hostAddressToTuple h)-isRoutable (SockAddrInet6 port _ h _) =-  port /= 0 && case hostAddress6ToTuple h of-    (0, 0, 0, 0, 0, 0xffff, hi, lo) ->-      routable4-        ( fromIntegral (hi `div` 256),-          fromIntegral (hi `mod` 256),-          fromIntegral (lo `div` 256),-          fromIntegral (lo `mod` 256)-        )-    (a, b, _, _, _, _, _, _)-      | a == 0 -> False -- unspecified, loopback, IPv4-compatible-      | a .&. 0xfe00 == 0xfc00 -> False -- unique local-      | a .&. 0xffc0 == 0xfe80 -> False -- link-local-      | a .&. 0xffc0 == 0xfec0 -> False -- site-local-      | a .&. 0xff00 == 0xff00 -> False -- multicast-      | a == 0x2001 && b == 0x0db8 -> False -- documentation-      | otherwise -> True-isRoutable _ = False---- | Whether an IPv4 address is public unicast.-routable4 :: (Word8, Word8, Word8, Word8) -> Bool-routable4 (a, b, c, _)-  | a == 0 = False -- this network-  | a == 10 = False -- private-  | a == 100 && b .&. 0xc0 == 64 = False -- shared address space-  | a == 127 = False -- loopback-  | a == 169 && b == 254 = False -- link-local-  | a == 172 && b .&. 0xf0 == 16 = False -- private-  | a == 192 && b == 0 && c == 0 = False -- protocol assignments-  | a == 192 && b == 0 && c == 2 = False -- documentation-  | a == 192 && b == 168 = False -- private-  | a == 198 && b .&. 0xfe == 18 = False -- benchmarking-  | a == 198 && b == 51 && c == 100 = False -- documentation-  | a == 203 && b == 0 && c == 113 = False -- documentation-  | a >= 224 = False -- multicast and reserved-  | otherwise = True---- | Split a peer address into host and port. Accepts @host@, @host:port@,--- @[ipv6]@ and @[ipv6]:port@. Either part is 'Nothing' if missing.-toHostService :: String -> (Maybe String, Maybe String)-toHostService str =-  let host = case m6 of-        Just (x, _) -> Just x-        Nothing -> case takeWhile (/= ':') str of-          [] -> Nothing-          xs -> Just xs-      srv = case m6 of-        Just (_, y) -> s y-        Nothing -> s str-      s xs =-        case dropWhile (/= ':') xs of-          [] -> Nothing-          _ : ys -> Just ys-      m6 = case str of-        (x : xs)-          | x == '[' -> do-              i <- elemIndex ']' xs-              return $ second (drop 1) $ splitAt i xs-          | x == ':' -> do-              return (str, "")-        _ -> Nothing-   in (host, srv)---- | Resolve a peer address given as in 'toHostService' into socket addresses,--- using the default port of the network if none is given. Returns an empty list--- if it cannot be resolved.-toSockAddr :: Network -> String -> IO [SockAddr]-toSockAddr net str =-  go `catch` e-  where-    go = fmap (map addrAddress) $ getAddrInfo Nothing host srv-    (host, srv) =-      second (<|> Just (show net.defaultPort)) $-        toHostService str-    e :: (Monad m) => SomeException -> m [SockAddr]-    e _ = return []  -- | Median of a list, or 'Nothing' if it is empty. median :: (Ord a, Fractional a) => [a] -> Maybe a

@@ -0,0 +1,208 @@+-- |+-- Module      : Haskoin.Node.PeerMgr.Addresses+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- The addresses that the peer manager connects to, as an explicit state+-- machine.+--+-- Every known address is 'Ready' to be tried, 'InUse' by a connection, or+-- 'Waiting' until a given time. How long an address waits after a connection+-- ends depends on how it ended ('Outcome').+--+-- = Why it works this way+--+-- * __Failing addresses back off.__ The node used to try a static peer that+--   refused connections again within seconds, forever, and an address it had+--   just disconnected for misbehaving could be connected again at once. Here a+--   failed address waits twice as long after each failure, up to+--   'maxBackoff', and a misbehaving one waits 'banTime'.+--+-- * __Configured peers are trusted more.__ Addresses from the configuration+--   are what the operator asked for, so they are never banned and never+--   forgotten: they back off at most 'maxBackoff', even after misbehaving.+--   Discovered addresses, from DNS seeds and gossip, are banned for+--   misbehaving, and at most 'maxDiscovered' of them are kept, so gossip+--   cannot exhaust memory.+--+-- * __Pure, with time as an input.__ Every transition takes the current time,+--   so the timing rules can be tested without waiting. "Haskoin.Node.PeerMgr"+--   keeps the book in a 'Control.Concurrent.STM.TVar'.+module Haskoin.Node.PeerMgr.Addresses+  ( AddressBook,+    Source (..),+    AddressState (..),+    Outcome (..),+    emptyBook,+    addAddress,+    takeAddress,+    returnAddress,+    markConnected,+    finishAddress,+    addressState,+    needsReload,+    markLoaded,+    knownAddresses,+    retryDelay,+    maxBackoff,+    banTime,+    reloadInterval,+    maxDiscovered,+  )+where++import Data.List (find)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Maybe (isNothing)+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime)++-- | Where an address came from.+data Source+  = -- | the static peers of the configuration+    Configured+  | -- | DNS seeds, or addresses relayed by peers+    Discovered+  deriving stock (Eq, Show)++-- | Whether an address can be tried.+data AddressState+  = -- | can be tried now+    Ready+  | -- | a connection to it is open or being opened+    InUse+  | -- | can be tried again at the given time; holds the number of failures in+    -- a row+    Waiting !UTCTime !Int+  deriving stock (Eq, Show)++-- | How a connection ended.+data Outcome+  = -- | the peer was online and the connection ended normally, such as by+    -- rotation or because the peer closed it+    Disconnected+  | -- | the connection could not be made, or the peer stopped responding+    Failed+  | -- | the peer broke the protocol+    Misbehaved+  deriving stock (Eq, Show)++data Entry = Entry+  { source :: !Source,+    state :: !AddressState,+    -- | failures in a row+    failures :: !Int+  }+  deriving stock (Eq, Show)++-- | Known addresses, and when the configured peers and seeds were last+-- loaded.+data AddressBook a = AddressBook+  { entries :: !(Map a Entry),+    loaded :: !(Maybe UTCTime)+  }+  deriving stock (Eq, Show)++-- | Wait after a normal disconnection before trying the address again.+retryDelay :: NominalDiffTime+retryDelay = 1++-- | Longest wait after failures.+maxBackoff :: NominalDiffTime+maxBackoff = 600++-- | Wait after a discovered address misbehaved.+banTime :: NominalDiffTime+banTime = 24 * 3600++-- | Shortest time between two loads of the configured peers and seeds.+reloadInterval :: NominalDiffTime+reloadInterval = 300++-- | Most discovered addresses kept.+maxDiscovered :: Int+maxDiscovered = 4096++-- | No addresses, never loaded.+emptyBook :: AddressBook a+emptyBook = AddressBook {entries = Map.empty, loaded = Nothing}++-- | Learn an address. A known address keeps its state; one from the+-- configuration is marked 'Configured'. A discovered address is dropped if+-- 'maxDiscovered' are known already.+addAddress :: (Ord a) => Source -> a -> AddressBook a -> AddressBook a+addAddress src a book =+  case Map.lookup a book.entries of+    Just e+      | src == Configured -> put e {source = Configured}+      | otherwise -> book+    Nothing+      | src == Discovered && discovered >= maxDiscovered -> book+      | otherwise -> put Entry {source = src, state = Ready, failures = 0}+  where+    put e = book {entries = Map.insert a e book.entries}+    discovered = Map.size (Map.filter ((== Discovered) . (.source)) book.entries)++-- | Take an address to connect to: a 'Ready' one, or a 'Waiting' one whose+-- time has come. It becomes 'InUse'.+takeAddress :: (Ord a) => UTCTime -> AddressBook a -> Maybe (a, AddressBook a)+takeAddress now book = do+  (a, e) <- find (ready . snd) (Map.toList book.entries)+  return (a, setEntry a e {state = InUse} book)+  where+    ready e = case e.state of+      Ready -> True+      InUse -> False+      Waiting t _ -> t <= now++-- | Give back an address taken with 'takeAddress' without connecting to it.+returnAddress :: (Ord a) => a -> AddressBook a -> AddressBook a+returnAddress = adjust $ \e -> case e.state of+  InUse -> e {state = Ready}+  _ -> e++-- | A connection to the address completed the handshake. Its failures are+-- forgotten.+markConnected :: (Ord a) => a -> AddressBook a -> AddressBook a+markConnected = adjust $ \e -> e {failures = 0}++-- | A connection to the address ended. Decide when to try it again.+finishAddress ::+  (Ord a) => UTCTime -> Outcome -> a -> AddressBook a -> AddressBook a+finishAddress now outcome = adjust $ \e ->+  let wait d f = e {state = Waiting (d `addUTCTime` now) f, failures = f}+      backoff f = min maxBackoff (retryDelay * 2 ^ min f 30)+   in case outcome of+        Disconnected -> wait retryDelay 0+        Failed -> let f = e.failures + 1 in wait (backoff f) f+        Misbehaved+          | e.source == Configured -> wait maxBackoff (e.failures + 1)+          | otherwise -> wait banTime (e.failures + 1)++-- | State of an address, or 'Nothing' if it is not known.+addressState :: (Ord a) => a -> AddressBook a -> Maybe AddressState+addressState a book = (.state) <$> Map.lookup a book.entries++-- | Whether to load the configured peers and seeds: when no address can be+-- tried now, and they were never loaded or not loaded for 'reloadInterval'.+needsReload :: (Ord a) => UTCTime -> AddressBook a -> Bool+needsReload now book =+  isNothing (takeAddress now book)+    && maybe True (\t -> now `diffUTCTime` t >= reloadInterval) book.loaded++-- | The configured peers and seeds were loaded at the given time.+markLoaded :: UTCTime -> AddressBook a -> AddressBook a+markLoaded now book = book {loaded = Just now}++-- | Every known address.+knownAddresses :: AddressBook a -> [a]+knownAddresses book = Map.keys book.entries++setEntry :: (Ord a) => a -> Entry -> AddressBook a -> AddressBook a+setEntry a e book = book {entries = Map.insert a e book.entries}++adjust :: (Ord a) => (Entry -> Entry) -> a -> AddressBook a -> AddressBook a+adjust f a book = book {entries = Map.adjust f a book.entries}

@@ -0,0 +1,142 @@+-- |+-- Module      : Haskoin.Node.PeerMgr.Handshake+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- The version handshake with a peer, as an explicit state machine.+--+-- The node sends its version message as soon as a connection opens. The peer+-- must answer with its own version and acknowledge ours with a verack, in+-- either order. Only once both have arrived is the peer usable, and only then+-- does "Haskoin.Node.PeerMgr" announce it with a+-- 'Haskoin.Node.Peer.PeerConnected' event.+--+-- = Why a state machine+--+-- The handshake used to be tracked with independent flags, one for the+-- version and one for the verack, with each message handler checking the+-- flags it cared about. Every handler had to remember every rule, and a+-- combination nobody thought of, such as a second verack, would silently do+-- something. Here the states are listed once, 'stepHandshake' says what each+-- message does in each state, and a message that makes no sense in the+-- current state is an error. Unforeseen sequences are rejected instead of+-- handled by accident, and the rules can be tested exhaustively without a+-- network.+--+-- The module is pure. "Haskoin.Node.PeerMgr" stores the state of each peer+-- and carries out the actions that a transition returns.+module Haskoin.Node.PeerMgr.Handshake+  ( Handshake (..),+    HandshakeEvent (..),+    HandshakeAction (..),+    HandshakeError (..),+    stepHandshake,+    handshakeVersion,+    isEstablished,+    hasVerAck,+    checkVersion,+  )+where++import Data.Bits ((.&.))+import Data.Word (Word64)+import Haskoin (Version (..), nodeNetwork)++-- | How far the handshake with a peer has progressed.+data Handshake+  = -- | Our version was sent; nothing has been received yet.+    AwaitingBoth+  | -- | The version of the peer arrived; its verack has not.+    AwaitingVerAck !Version+  | -- | The verack of the peer arrived; its version has not.+    AwaitingVersion+  | -- | Both arrived. The peer is online.+    Established !Version+  deriving stock (Eq, Show)++-- | Handshake messages received from a peer.+data HandshakeEvent+  = -- | the peer sent its version+    GotVersion !Version+  | -- | the peer acknowledged our version+    GotVerAck+  deriving stock (Eq, Show)++-- | What the peer manager must do after a transition.+data HandshakeAction+  = -- | acknowledge the version of the peer with a verack+    SendVerAck+  | -- | publish that the peer is online+    AnnounceConnected+  deriving stock (Eq, Show)++-- | Reasons to disconnect a peer during the handshake.+data HandshakeError+  = -- | the peer sent a second version message+    DuplicateVersion+  | -- | the peer sent a second verack+    DuplicateVerAck+  | -- | the peer does not serve the full block chain+    NotFullNode+  | -- | the version carries one of our own nonces: we connected to ourselves+    ConnectedToSelf+  deriving stock (Eq, Show)++-- | Apply a message to the handshake. Returns the new state and the actions to+-- carry out, or why the peer must be disconnected.+--+-- Each message may arrive once. A second version or verack is a protocol+-- violation: well-behaved peers never send one, and accepting them would let+-- a peer announce itself as connected twice.+stepHandshake ::+  Handshake ->+  HandshakeEvent ->+  Either HandshakeError (Handshake, [HandshakeAction])+stepHandshake state event =+  case (state, event) of+    (AwaitingBoth, GotVersion v) -> Right (AwaitingVerAck v, [SendVerAck])+    (AwaitingBoth, GotVerAck) -> Right (AwaitingVersion, [])+    (AwaitingVerAck v, GotVerAck) -> Right (Established v, [AnnounceConnected])+    (AwaitingVersion, GotVersion v) ->+      Right (Established v, [SendVerAck, AnnounceConnected])+    (AwaitingVerAck _, GotVersion _) -> Left DuplicateVersion+    (AwaitingVersion, GotVerAck) -> Left DuplicateVerAck+    (Established _, GotVersion _) -> Left DuplicateVersion+    (Established _, GotVerAck) -> Left DuplicateVerAck++-- | The version of the peer, once it has arrived.+handshakeVersion :: Handshake -> Maybe Version+handshakeVersion = \case+  AwaitingVerAck v -> Just v+  Established v -> Just v+  AwaitingBoth -> Nothing+  AwaitingVersion -> Nothing++-- | Whether the handshake is complete.+isEstablished :: Handshake -> Bool+isEstablished = \case+  Established _ -> True+  _ -> False++-- | Whether the peer acknowledged our version.+hasVerAck :: Handshake -> Bool+hasVerAck = \case+  AwaitingVersion -> True+  Established _ -> True+  _ -> False++-- | Check the version of a peer before accepting it. Takes the nonces of the+-- version messages that we sent to all our peers.+--+-- A peer must serve the full chain ('nodeNetwork'), because the node syncs+-- headers and applications download blocks from it. A version carrying one of+-- our nonces means the node connected to itself, which happens when it learns+-- its own address from gossip.+checkVersion :: [Word64] -> Version -> Either HandshakeError Version+checkVersion nonces v+  | v.services .&. nodeNetwork == 0 = Left NotFullNode+  | v.nonce `elem` nonces = Left ConnectedToSelf+  | otherwise = Right v

@@ -0,0 +1,209 @@+-- |+-- Module      : Haskoin.Node.PeerMgr.Lifecycle+-- Copyright   : (c) Haskoin Developers+-- License     : MIT+-- Maintainer  : jprupp@protonmail.ch+-- Stability   : experimental+-- Portability : POSIX+--+-- The life of a connection to a peer, as an explicit state machine: the+-- handshake, then keeping the peer alive with pings, then disconnecting.+--+-- "Haskoin.Node.PeerMgr" keeps a 'PeerState' for every connection. It feeds+-- 'stepPeer' what happens (a message arrived, a pong came back, the clock+-- ticked) and carries out the 'PeerAction's that come back.+--+-- = Why it works this way+--+-- * __Every stage is named.__ A connection is 'Handshaking', 'Online' or+--   'Disconnecting'. Before, a peer that had been told to disconnect looked+--   exactly like an online one until its process died: it was still offered+--   to applications, still had its messages routed, and could be told to+--   disconnect again. 'Disconnecting' makes that window explicit, and every+--   input in it is ignored.+--+-- * __Time is an input.__ The rules about silence, pings and age depend on+--   the clock. Instead of reading it, 'stepPeer' takes the time in 'Tick',+--   'Heard' and 'GotPong'. Every timing rule can then be tested by passing+--   times, without waiting.+--+-- * __Randomness is an input.__ A ping needs a fresh nonce. 'Tick' carries+--   one, so that the transition stays a pure function.+--+-- * __One reason per disconnection.__ 'DisconnectReason' records why the node+--   gave up on a peer, so that the peer manager can decide whether to try the+--   address again soon, later, or not for a long time (see+--   "Haskoin.Node.PeerMgr.Addresses").+module Haskoin.Node.PeerMgr.Lifecycle+  ( PeerState (..),+    Stage (..),+    Limits (..),+    PeerInput (..),+    PeerAction (..),+    DisconnectReason (..),+    newPeerState,+    stepPeer,+    isOnline,+    stageVersion,+    pendingPing,+  )+where++import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime)+import Data.Word (Word64)+import Haskoin (Version)+import Haskoin.Node.PeerMgr.Handshake++-- | Stage of a connection.+data Stage+  = -- | the version handshake is in progress+    Handshaking !Handshake+  | -- | the handshake is complete; holds the version of the peer and the ping+    -- awaiting an answer, with the time it was sent and its nonce+    Online !Version !(Maybe (UTCTime, Word64))+  | -- | the peer was told to disconnect, for the given reason+    Disconnecting !DisconnectReason+  deriving stock (Eq, Show)++-- | Why the node disconnects from a peer.+data DisconnectReason+  = -- | the peer broke the handshake rules, or its version was refused+    Rejected !HandshakeError+  | -- | the handshake did not complete in time+    HandshakeTimeout+  | -- | the peer sent nothing for too long+    Silent+  | -- | the peer did not answer a ping in time+    NoPong+  | -- | the peer has been connected for the maximum time; connections rotate+    TooOld+  deriving stock (Eq, Show)++-- | State of a connection.+data PeerState = PeerState+  { -- | stage of the connection+    stage :: !Stage,+    -- | when the connection was opened+    connected :: !UTCTime,+    -- | when the peer last sent a message+    lastHeard :: !UTCTime,+    -- | round-trip times of the last eleven pings, newest first+    pings :: ![NominalDiffTime],+    -- | whether the peer was announced as connected, so that its+    -- disconnection must be announced too+    announced :: !Bool+  }+  deriving stock (Eq, Show)++-- | Time limits for a connection.+data Limits = Limits+  { -- | longest silence, unanswered ping or handshake+    timeout :: !NominalDiffTime,+    -- | longest connection, after which it is rotated+    maxLife :: !NominalDiffTime+  }+  deriving stock (Eq, Show)++-- | What happened to a connection.+data PeerInput+  = -- | the peer sent a message at the given time+    Heard !UTCTime+  | -- | the peer sent a handshake message+    GotHandshake !HandshakeEvent+  | -- | the version of the peer was refused (see 'checkVersion')+    Refused !HandshakeError+  | -- | the peer answered a ping with the given nonce at the given time+    GotPong !UTCTime !Word64+  | -- | the clock ticked; holds the time, and a nonce for a ping if one is due+    Tick !UTCTime !Word64+  deriving stock (Eq, Show)++-- | What the peer manager must do after a transition.+data PeerAction+  = -- | acknowledge the version of the peer+    ReplyVerAck+  | -- | publish that the peer is online+    Announce+  | -- | send a ping with the given nonce+    SendPing !Word64+  | -- | tell the peer process to disconnect+    Disconnect !DisconnectReason+  deriving stock (Eq, Show)++-- | A connection opened at the given time, with our version sent.+newPeerState :: UTCTime -> PeerState+newPeerState now =+  PeerState+    { stage = Handshaking AwaitingBoth,+      connected = now,+      lastHeard = now,+      pings = [],+      announced = False+    }++-- | Apply an input to a connection.+--+-- On a 'Tick', the checks come in this order: the connection is too old, the+-- peer has been silent too long, the handshake took too long, a ping went+-- unanswered too long. If none applies and no ping is pending, a ping is sent.+stepPeer :: Limits -> PeerInput -> PeerState -> (PeerState, [PeerAction])+stepPeer limits input st =+  case (st.stage, input) of+    (Disconnecting _, _) -> (st, [])+    (_, Heard t) -> (st {lastHeard = t}, [])+    (_, Refused e) -> disconnect (Rejected e)+    (Handshaking h, GotHandshake e) -> handshake h e+    (Online _ _, GotHandshake (GotVersion _)) -> disconnect (Rejected DuplicateVersion)+    (Online _ _, GotHandshake GotVerAck) -> disconnect (Rejected DuplicateVerAck)+    (Online v (Just (sent, n)), GotPong t n')+      | n == n' ->+          ( st+              { stage = Online v Nothing,+                pings = take 11 (t `diffUTCTime` sent : st.pings)+              },+            []+          )+    (_, GotPong _ _) -> (st, [])+    (stage, Tick now nonce)+      | now `since` st.connected > limits.maxLife -> disconnect TooOld+      | now `since` st.lastHeard > limits.timeout -> disconnect Silent+      | Handshaking _ <- stage,+        now `since` st.connected > limits.timeout ->+          disconnect HandshakeTimeout+      | Online _ (Just (sent, _)) <- stage,+        now `since` sent > limits.timeout ->+          disconnect NoPong+      | Online v Nothing <- stage ->+          (st {stage = Online v (Just (now, nonce))}, [SendPing nonce])+      | otherwise -> (st, [])+  where+    since = diffUTCTime+    disconnect r = (st {stage = Disconnecting r}, [Disconnect r])+    handshake h e =+      case stepHandshake h e of+        Left err -> disconnect (Rejected err)+        Right (Established v, acts) ->+          (st {stage = Online v Nothing, announced = True}, map action acts)+        Right (h', acts) -> (st {stage = Handshaking h'}, map action acts)+    action = \case+      SendVerAck -> ReplyVerAck+      AnnounceConnected -> Announce++-- | Whether the handshake is complete and the peer is not being disconnected.+isOnline :: PeerState -> Bool+isOnline st = case st.stage of+  Online _ _ -> True+  _ -> False++-- | Version of the peer, once received.+stageVersion :: PeerState -> Maybe Version+stageVersion st = case st.stage of+  Handshaking h -> handshakeVersion h+  Online v _ -> Just v+  Disconnecting _ -> Nothing++-- | The ping awaiting an answer: when it was sent, and its nonce.+pendingPing :: PeerState -> Maybe (UTCTime, Word64)+pendingPing st = case st.stage of+  Online _ p -> p+  _ -> Nothing

@@ -1,552 +0,0 @@-{-# LANGUAGE DuplicateRecordFields #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ImportQualifiedPost #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedRecordDot #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TupleSections #-}-{-# LANGUAGE NoFieldSelectors #-}--module Haskoin.NodeSpec (spec) where--import Conduit-import Control.Monad (forM_, forever, replicateM)-import Control.Monad.Cont-import Control.Monad.Trans (lift)-import Data.ByteString (ByteString)-import Data.ByteString qualified as B-import Data.ByteString.Base64 (decodeBase64Lenient)-import Data.Default (def)-import Data.Either (fromRight)-import Data.List (find)-import Data.Maybe (isJust, listToMaybe, mapMaybe)-import Data.Serialize (decode, get, put, runGet, runPut)-import Data.Time.Clock (NominalDiffTime)-import Data.Time.Clock.POSIX (getPOSIXTime)-import Database.RocksDB qualified as R-import Haskoin-import Haskoin.Node-import NQE-import Network.Socket (AddrInfo (addrAddress), SockAddr (..), tupleToHostAddress, tupleToHostAddress6)-import System.Random (randomIO)-import Test.Hspec-import Test.Hspec.QuickCheck-import UnliftIO-import UnliftIO.Concurrent (threadDelay)--data TestNode = TestNode-  { testMgr :: PeerMgr,-    testChain :: Chain,-    nodeEvents :: Inbox NodeEvent,-    testDB :: R.DB,-    -- | how many getheaders messages the mock peer has received-    getHeadersCount :: TVar Int-  }---- | What the mock peer sends back for each message: a message to serialize--- for the test network, or raw bytes.-type React = Message -> [Either ByteString Message]---- | How to set up a test node.-data TestOpts = TestOpts-  { react :: React,-    -- | runs on the database before the node starts-    setup :: R.DB -> IO (),-    timeout :: NominalDiffTime-  }--defOpts :: [Block] -> TestOpts-defOpts blocks =-  TestOpts-    { react = map Right . mockPeerReact blocks,-      setup = \_ -> return (),-      timeout = 120-    }---- | Connect to a mock peer that reacts to messages like this.-dummyPeerConnect ::-  React ->-  TVar Int ->-  Network ->-  NetworkAddress ->-  SockAddr ->-  WithConnection-dummyPeerConnect react count net ad sa f = do-  r <- newInbox-  s <- newInbox-  let s' = inboxToMailbox s-  withAsync (go r s') $ \_ -> do-    let o = awaitForever (`send` r)-        i = forever (receive s >>= yield)-    f (Conduits i o) :: IO ()-  where-    go :: Inbox ByteString -> Mailbox ByteString -> IO ()-    go r s = do-      nonce <- randomIO-      now <- round <$> getPOSIXTime-      let rmt = NetworkAddress 0 (sockToHostAddress sa)-          ver = buildVersion net nonce 0 ad rmt now-      runPut (putMessage net (MVersion ver)) `send` s-      runConduit $-        forever (receive r >>= yield)-          .| inc-          .| iterMC count_getheaders-          .| concatMapC react-          .| outc-          .| awaitForever (`send` s)-    outc = mapMC $ return . either id (runPut . putMessage net)-    count_getheaders = \case-      MGetHeaders _ -> atomically (modifyTVar' count (+ 1))-      _ -> return ()-    inc =-      forever $ do-        x <- takeCE 24 .| foldC-        y <- case decode x of-          Left _ -> error "Dummy peer not decode message header"-          Right (MessageHeader _ _ len _) ->-            takeCE (fromIntegral len) .| foldC-        case runGet (getMessage net) $ x `B.append` y of-          Left e ->-            error $-              "Dummy peer could not decode payload: " <> show e-          Right msg -> yield msg--mockPeerReact :: [Block] -> Message -> [Message]-mockPeerReact _ (MPing (Ping n)) = [MPong (Pong n)]-mockPeerReact _ (MVersion _) = [MVerAck]-mockPeerReact blocks (MGetHeaders (GetHeaders _ _hs _)) = [MHeaders (Headers hs')]-  where-    f b = (b.header, (VarInt . fromIntegral . length) b.txs)-    hs' = map f blocks-mockPeerReact blocks (MGetData (GetData ivs)) = mapMaybe f ivs-  where-    f (InvVector InvBlock h) = MBlock <$> find (l h) blocks-    f _ = Nothing-    l h b = headerHash b.header == BlockHash h-mockPeerReact _ _ = []--spec :: Spec-spec = do-  let net = bchRegTest-  describe "reads address/port combinations" $ do-    -- IPv4 only: resolving an IPv6 address needs IPv6 on the host, which-    -- containers and CI runners often lack. The IPv6 forms are still parsed-    -- by the toHostService tests below, which need no resolver.-    prop "reads arbitrary IPv4 addresses" $ \(e, w) -> do-      let p = toEnum (e `mod` 65536)-          a = SockAddrInet p w-      s <- toSockAddr net (show a)-      s `shouldStartWith` [a]-    it "reads some specific addresses" $ do-      toHostService "localhost" `shouldBe` (Just "localhost", Nothing)-      toHostService "::1" `shouldBe` (Just "::1", Nothing)-      toHostService "localhost:8080" `shouldBe` (Just "localhost", Just "8080")-      toHostService "example.com" `shouldBe` (Just "example.com", Nothing)-      toHostService "api.example.com:443" `shouldBe` (Just "api.example.com", Just "443")-      toHostService "api.example.com:http" `shouldBe` (Just "api.example.com", Just "http")-      toHostService "[::1]" `shouldBe` (Just "::1", Nothing)-      toHostService "[::1]:8080" `shouldBe` (Just "::1", Just "8080")-      toHostService "[2002::dead:beef]:ssh" `shouldBe` (Just "2002::dead:beef", Just "ssh")-  describe "peer manager on test network" $ do-    it "connects to a peer" $-      withTestNode net "connect-one-peer" $ \TestNode {..} -> do-        p <- waitForPeer nodeEvents-        Just OnlinePeer {version = Just Version {version = ver}} <--          getOnlinePeer testMgr p-        ver `shouldSatisfy` (>= 70002)-    it "downloads some blocks" $-      withTestNode net "get-blocks" $ \TestNode {..} -> do-        let h1 =-              "3094ed3592a06f3d8e099eed2d9c1192329944f5df4a48acb29e08f12cfbb660"-            h2 =-              "0c89955fc5c9f98ecc71954f167b938138c90c6a094c4737f2e901669d26763f"-        p <- waitForPeer nodeEvents-        pbs <- getBlocks net 10 p [h1, h2]-        pbs `shouldSatisfy` isJust-        let Just [b1, b2] = pbs-        headerHash b1.header `shouldBe` h1-        headerHash b2.header `shouldBe` h2-        let ths b = map txHash b.txs-            testMerkle b = buildMerkleRoot (ths b) `shouldBe` Right b.header.merkle-        testMerkle b1-        testMerkle b2-    it "downloads a block larger than 2 MiB" $ do-      -- On Bitcoin Cash a block message may be up to twice the block size,-      -- while every other message is limited to 2 MiB (2097152 bytes).-      let script = B.replicate 3000000 0x6a-          tx = Tx 1 [TxIn nullOutPoint B.empty maxBound] [TxOut 0 script] [] 0-          firstBlock : _ = allBlocks-          big = Block firstBlock.header [tx]-      withTestNodeBlocks [big] net "big-block" $ \TestNode {..} -> do-        p <- waitForPeer nodeEvents-        pbs <- getBlocks net 10 p [headerHash big.header]-        let scriptSizes =-              [[B.length o.script | o <- t.outputs] | b <- concat pbs, t <- b.txs]-        scriptSizes `shouldBe` [[3000000]]-    it "builds a version message with the given height and time" $ do-      let local = NetworkAddress nodeNetwork (sockToHostAddress (SockAddrInet 0 0))-          remote = NetworkAddress 0 (sockToHostAddress (SockAddrInet 8333 0x08080808))-          v = buildVersion net 7 650000 local remote 1700000000-      v.version `shouldBe` 70012-      v.startHeight `shouldBe` 650000-      v.timestamp `shouldBe` 1700000000-  describe "untrusted peers" $ do-    it "imports a full batch of 2000 headers" $-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Right (manyHeaders net 2000)]}-        net-        "full-batch"-        $ \TestNode {..} -> do-          bn <--            receiveMatch nodeEvents $ \case-              ChainEvent (ChainBestBlock bn) | bn.height > 0 -> Just bn-              _ -> Nothing-          bn.height `shouldBe` 2000-    it "disconnects a peer that sends more than 2000 headers" $-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Right (manyHeaders net 2001)]}-        net-        "too-many-headers"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-          (.height) <$> chainGetBest testChain `shouldReturn` 0-    it "disconnects a peer that sends more than 1000 addresses" $ do-      let na = NetworkAddress 0 (sockToHostAddress (SockAddrInet 8333 0x08080808))-          addrs = MAddr (Addr (replicate 1001 (0, na)))-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Right addrs]}-        net-        "too-many-addrs"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-    it "disconnects a peer that sends the wrong network magic" $ do-      let bad = runPut (putMessage btc (MPing (Ping 1)))-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Left bad]}-        net-        "wrong-magic"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-    it "disconnects a peer that announces a tx message larger than 2 MiB" $ do-      -- Only the header is sent, with a length of 2 MiB plus one byte:-      -- 2 × 1024 × 1024 + 1 = 2097153. The node must refuse it there,-      -- without waiting for the payload.-      let oversized =-            runPut . put $-              MessageHeader net.magic MCTx 2097153 (checkSum32 B.empty)-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Left oversized]}-        net-        "oversized-tx"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-    it "disconnects a peer that sends a second verack" $-      withTestNodeOpts-        (defOpts allBlocks) {react = reactAfterVerAck [Right MVerAck]}-        net-        "duplicate-verack"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-    it "ignores headers sent before the handshake" $ do-      let hs = MHeaders (Headers [(b.header, VarInt 0) | b <- allBlocks])-          react = \case-            MVersion _ -> [Right hs, Right MVerAck]-            MPing (Ping n) -> [Right (MPong (Pong n))]-            _ -> []-      withTestNodeOpts (defOpts allBlocks) {react = react} net "early-headers" $-        \TestNode {..} -> do-          _ <- waitForPeer nodeEvents-          threadDelay (500 * 1000)-          (.height) <$> chainGetBest testChain `shouldReturn` 0-    it "disconnects a peer that chatters instead of answering a ping" $ do-      -- Answer the node's ping with a ping of our own, and its pong with-      -- another ping, so that the peer is never silent.-      let react = \case-            MVersion _ -> [Right MVerAck]-            MPing _ -> [Right (MPing (Ping 1))]-            MPong _ -> [Right (MPing (Ping 1))]-            -- end the header sync so its own timeout does not apply-            MGetHeaders _ -> [Right (MHeaders (Headers []))]-            _ -> []-      withTestNodeOpts-        (defOpts allBlocks) {react = react, timeout = 2}-        net-        "ignored-ping"-        $ \TestNode {..} -> do-          p <- waitForPeer nodeEvents-          waitForDisconnect nodeEvents p-    it "does not connect to private addresses that peers relay" $ do-      let v4 a b c d p = SockAddrInet p (tupleToHostAddress (a, b, c, d))-          v6 t p = SockAddrInet6 p 0 (tupleToHostAddress6 t) 0-      isRoutable (v4 8 8 8 8 8333) `shouldBe` True-      isRoutable (v4 8 8 8 8 0) `shouldBe` False-      isRoutable (v4 127 0 0 1 8333) `shouldBe` False-      isRoutable (v4 10 1 2 3 8333) `shouldBe` False-      isRoutable (v4 172 16 0 1 8333) `shouldBe` False-      isRoutable (v4 172 32 0 1 8333) `shouldBe` True-      isRoutable (v4 192 168 1 1 8333) `shouldBe` False-      isRoutable (v4 169 254 1 1 8333) `shouldBe` False-      isRoutable (v4 100 64 0 1 8333) `shouldBe` False-      isRoutable (v4 224 0 0 1 8333) `shouldBe` False-      isRoutable (v6 (0x2a01, 0x4f8, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` True-      isRoutable (v6 (0, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False-      isRoutable (v6 (0xfd00, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False-      isRoutable (v6 (0xfe80, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False-      isRoutable (v6 (0, 0, 0, 0, 0, 0xffff, 0x7f00, 1) 8333) `shouldBe` False-      isRoutable (v6 (0, 0, 0, 0, 0, 0xffff, 0x0808, 0x0808) 8333) `shouldBe` True-      isRoutable (hostToSockAddr (sockToHostAddress (v4 127 0 0 1 8333)))-        `shouldBe` False-  describe "chain database" $ do-    it "leaves the default column family alone when purging" $ do-      -- The same key in both families: the purge must delete it only from-      -- the family that holds the chain.-      let k = B.pack [0x90, 1, 2, 3]-          setup db = do-            R.put db k "keep"-            forM_ (R.columnFamilies db) $ \cf -> R.putCF db cf k "stale"-      withTestNodeOpts (defOpts allBlocks) {setup = setup} net "purge-cf" $-        \TestNode {..} -> do-          R.get testDB k `shouldReturn` Just "keep"-  describe "chain on test network" $ do-    it "stops asking a peer that has no more headers" $-      -- The peer answers getheaders with an empty headers message, as a-      -- peer does when the node already has its whole chain. The node must-      -- take that as the end of the sync instead of asking again.-      withTestNodeBlocks [] net "no-headers" $ \TestNode {..} -> do-        _ <- waitForPeer nodeEvents-        atomically $ readTVar getHeadersCount >>= checkSTM . (>= 1)-        threadDelay (1000 * 1000)-        readTVarIO getHeadersCount `shouldReturn` 1-    it "reports the headers as synced when the best header has the minimum chain work" $ do-      -- Every regtest header has a work of 2. Genesis and one header on top-      -- of it have 2 + 2 = 4, which is the minimum here.-      fresh <- freshHeader net-      let net4 = net {minimumChainWork = 4}-      withTestNodeBlocks [Block fresh []] net4 "min-work-reached" $-        \TestNode {..} -> do-          receiveMatch nodeEvents $ \case-            ChainEvent (ChainBestBlock bn) | bn.height == 1 -> Just ()-            _ -> Nothing-          chainSync testChain-          chainIsSynced testChain `shouldReturn` True-    it "does not report the headers as synced below the minimum chain work" $ do-      -- Same chain with a work of 2 + 2 = 4, one short of a minimum of 5.-      fresh <- freshHeader net-      let net5 = net {minimumChainWork = 5}-      withTestNodeBlocks [Block fresh []] net5 "min-work-missing" $-        \TestNode {..} -> do-          receiveMatch nodeEvents $ \case-            ChainEvent (ChainBestBlock bn) | bn.height == 1 -> Just ()-            _ -> Nothing-          chainSync testChain-          chainIsSynced testChain `shouldReturn` False-    it "syncs some headers" $-      withTestNode net "connect-sync" $ \TestNode {..} -> do-        let bh =-              "3bfa0c6da615fc45aa44ddea6854ac19d16f3ca167e0e21ac2cc262a49c9b002"-            ah =-              "7dc835a78a55fa76f9184dc4f6663a73e418c7afec789c5ae25e432fd7fc8467"-        bn <--          receiveMatch nodeEvents $ \case-            ChainEvent (ChainBestBlock bn)-              | bn.height > 0 -> Just bn-            _ -> Nothing-        bb <- chainGetBest testChain-        bb.height `shouldSatisfy` (== 15)-        an <--          maybe (error "No ancestor found") return-            =<< chainGetAncestor testChain 10 bn-        headerHash bn.header `shouldBe` bh-        headerHash an.header `shouldBe` ah-    it "downloads some block parents" $-      withTestNode net "parents" $ \TestNode {..} -> do-        let hs =-              [ "52e886df7b166d961ac2d3d2d561d806325d51a609dc0a5d9d5fcb65d47906d7",-                "2537a081b9e2b24d217fac2886f387758cb3aa4e4956b3be7ed229bafbb71b0f",-                "7c72f306215a296f9714320a497b1f2cb5f9b99f162d7e04333c243fac9a54d8"-              ]-        [_, bn] <--          replicateM 2 $-            receiveMatch nodeEvents $ \case-              ChainEvent (ChainBestBlock bn) -> Just bn-              _ -> Nothing-        bn.height `shouldBe` 15-        ps <- chainGetParents testChain 12 bn-        length ps `shouldBe` 3-        forM_ (zip ps hs) $ \(p, h) ->-          headerHash p.header `shouldBe` h---- | Wait for the node to disconnect from this peer, failing after a while.-waitForDisconnect :: Inbox NodeEvent -> Peer -> IO ()-waitForDisconnect inbox p =-  timeout (20 * 1000 * 1000) wait >>= \case-    Just () -> return ()-    Nothing -> expectationFailure "peer was not disconnected"-  where-    wait =-      receiveMatch inbox $ \case-        PeerEvent (PeerDisconnected p') | p == p' -> Just ()-        _ -> Nothing---- | Mock peer that answers the handshake and pings, and sends these right--- after its version ack.-reactAfterVerAck :: [Either ByteString Message] -> React-reactAfterVerAck extra = \case-  MVersion _ -> Right MVerAck : extra-  MPing (Ping n) -> [Right (MPong (Pong n))]-  _ -> []---- | Headers message with a valid chain of this many headers on top of--- genesis.-manyHeaders :: Network -> Int -> Message-manyHeaders net n =-  MHeaders . Headers $-    map (,VarInt 0) (appendBlocks net 0 net.genesisHeader n)---- | Valid header on top of genesis that is dated now. The node only counts--- its headers as synced when the best one is less than two hours old, and--- the headers of the other fixtures are years old.-freshHeader :: Network -> IO BlockHeader-freshHeader net = do-  now <- round <$> getPOSIXTime-  let genesis = net.genesisHeader-  return $-    mineBlock-      net-      0-      genesis {version = 4, prev = headerHash genesis, timestamp = now}--waitForPeer :: (MonadIO m) => Inbox NodeEvent -> m Peer-waitForPeer inbox =-  receiveMatch inbox $ \case-    PeerEvent (PeerConnected p) -> Just p-    _ -> Nothing--withLifted :: (MonadUnliftIO m) => ((a -> IO b) -> IO b) -> (a -> m b) -> m b-withLifted w m = withRunInIO $ \io -> w (io . m)--withTestNode ::-  (MonadUnliftIO m) =>-  Network -> String -> (TestNode -> m a) -> m a-withTestNode = withTestNodeBlocks allBlocks---- | Like 'withTestNode', with the mock peer serving these blocks.-withTestNodeBlocks ::-  (MonadUnliftIO m) =>-  [Block] -> Network -> String -> (TestNode -> m a) -> m a-withTestNodeBlocks blocks = withTestNodeOpts (defOpts blocks)--withTestNodeOpts ::-  (MonadUnliftIO m) =>-  TestOpts -> Network -> String -> (TestNode -> m a) -> m a-withTestNodeOpts opts net str f = flip runContT return $ do-  w <- ContT $ withSystemTempDirectory ("haskoin-node-test-" <> str <> "-")-  count <- newTVarIO 0-  pub <- ContT withPublisher-  sub <- ContT $ withSubscription pub-  db <- ContT $ withLifted (R.withDBCF w cfg cols)-  liftIO (opts.setup db)-  let ad =-        NetworkAddress-          nodeNetwork-          (sockToHostAddress (SockAddrInet 0 0))-      na =-        NetworkAddress-          0-          (sockToHostAddress (SockAddrInet 0 0))-      cfg' =-        NodeConfig-          { maxPeers = 20,-            db = db,-            cf = listToMaybe (R.columnFamilies db),-            -- IPv4, so that the mock peer is reachable without IPv6-            peers = ["127.0.0.1:17486"],-            discover = False,-            address = na,-            net = net,-            pub = pub,-            timeout = opts.timeout,-            maxPeerLife = 48 * 3600,-            connect = dummyPeerConnect opts.react count net ad,-            logger = Logger {level = LevelError, source = "", push = \_ -> return ()}-          }-  Node mgr ch <- ContT $ withNode cfg'-  lift $-    f-      TestNode-        { testMgr = mgr,-          testChain = ch,-          nodeEvents = sub,-          testDB = db,-          getHeadersCount = count-        }-  where-    cfg = def {R.createIfMissing = True, R.errorIfExists = True}-    cols = [("node", def)]--allBlocks :: [Block]-allBlocks =-  fromRight (error "Could not decode blocks") $-    runGet f (decodeBase64Lenient allBlocksBase64)-  where-    f = mapM (const get) [(1 :: Int) .. 15]--allBlocksBase64 :: ByteString-allBlocksBase64 =-  "AAAAIAYibkYRGgtZyq8SYEPrW78ow086XjMqH8eytzzxiJEPakRJalmWTFwdvzNuH8fHLZEjn+4N\-  \FNMANdB7ez2M4a3TFbNe//9/IAMAAAABAgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\-  \AAAAAP////8MUQEBCC9FQjMyLjAv/////wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf\-  \1OHjgkqxnwEYkK9zrAAAAAAAAAAge0RDjOrqVayGUoQsbNTJcTXUM+psaHpmuiFy6hwo2T8yn0CL\-  \7WDJw9hxl1kf5c4JySq3WJF8OPsoguzF7mXH3tQVs17//38gAAAAAAECAAAAAQAAAAAAAAAAAAAA\-  \AAAAAAAAAAAAAAAAAAAAAAAAAAAA/////wxSAQEIL0VCMzIuMC//////AQDyBSoBAAAAIyEDBOym\-  \QKMx7MqzjsE+lp+i7WOOwd/U4eOCSrGfARiQr3OsAAAAAAAAACCKlhzDaFkrsmO2FhmeQS9ONS8D\-  \QsU4H97yNxVhyIXYJuG3a9cyQpdeETjCQ6JybgkwI0OOfa4eYazf7WWI5UAk1BWzXv//fyAEAAAA\-  \AQIAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD/////DFMBAQgvRUIzMi4wL///\-  \//8BAPIFKgEAAAAjIQME7KZAozHsyrOOwT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAAAAAAIP/S\-  \XiIJZqvUyBY90z72dv6+/GG50R3vc3UAK8AHP89wChmkVP6nefjOt+sNyhbKk9zia47F08oTNtC0\-  \OG1zyuXVFbNe//9/IAEAAAABAgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP//\-  \//8MVAEBCC9FQjMyLjAv/////wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf1OHjgkqx\-  \nwEYkK9zrAAAAAAAAAAgeQtE1s3YV/uS2jUouo3S9DJAVf5OGk+Nyx+No1mPH24b5JCkr/tSP0E/\-  \NYVkVcE0ZHxbO/fu5wOd+8VolvPQYtUVs17//38gAAAAAAECAAAAAQAAAAAAAAAAAAAAAAAAAAAA\-  \AAAAAAAAAAAAAAAAAAAA/////wxVAQEIL0VCMzIuMC//////AQDyBSoBAAAAIyEDBOymQKMx7Mqz\-  \jsE+lp+i7WOOwd/U4eOCSrGfARiQr3OsAAAAAAAAACBgtvss8QiesqxISt/1RJkykhGcLe2eCY49\-  \b6CSNe2UMOVYGZ++uRCKvaJ2+jo7akr7XsdXCYSAmuw6DwSO8lvF1RWzXv//fyAAAAAAAQIAAAAB\-  \AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD/////DFYBAQgvRUIzMi4wL/////8BAPIF\-  \KgEAAAAjIQME7KZAozHsyrOOwT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAAAAAAID92Jp1mAeny\-  \N0dMCWoMyTiBk3sWT5VxzI75ycVflYkMCnXLFhuwrMdBbZmXJinAJBUpN7BV0XvlM2PRmb7HQebV\-  \FbNe//9/IAEAAAABAgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP////8MVwEB\-  \CC9FQjMyLjAv/////wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf1OHjgkqxnwEYkK9z\-  \rAAAAAAAAAAgxEgEkhjf5p+ql8dETmdSCdCdk+vB26+V2SGLEuE1+kA1acGCdQoQBqec8P/knItJ\-  \M213OIrDX6U5IB6fgIas7dYVs17//38gAQAAAAECAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\-  \AAAAAAAAAAAA/////wxYAQEIL0VCMzIuMC//////AQDyBSoBAAAAIyEDBOymQKMx7MqzjsE+lp+i\-  \7WOOwd/U4eOCSrGfARiQr3OsAAAAAAAAACDku4EB5X7htWpHg+aMzzW1AABttpNQTew7K3Aj2fh/\-  \OuOCPhJApmcXq5o42tkksFSuhYvcfqaSHCuuFgjo6ohz1hWzXv//fyAAAAAAAQIAAAABAAAAAAAA\-  \AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD/////DFkBAQgvRUIzMi4wL/////8BAPIFKgEAAAAj\-  \IQME7KZAozHsyrOOwT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAAAAAAIKWpAhOWbkEN9vWf1uCu\-  \eXtVOZIE9V1OE87iC+H9atBRtY4LPgaWUSVMNh9SeZK1NViIFMklbjsfqYiC4eA/VuLWFbNe//9/\-  \IAAAAAABAgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP////8MWgEBCC9FQjMy\-  \LjAv/////wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf1OHjgkqxnwEYkK9zrAAAAAAA\-  \AAAgZ4T81y9DXuJanHjsr8cY5HM6ZvbETRj5dvpViqc1yH0oN9OOruaO5mjdITJwweVCzjSQ5Wsl\-  \vSOKaKvEX5j9l9YVs17//38gAAAAAAECAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\-  \AAAA/////wxbAQEIL0VCMzIuMC//////AQDyBSoBAAAAIyEDBOymQKMx7MqzjsE+lp+i7WOOwd/U\-  \4eOCSrGfARiQr3OsAAAAAAAAACCV3J2A3qneSJ7Q/RuF8OPd8O1izIXvKElR/xg/+InGNEafu0Ul\-  \3VYJR93zbAQuns9hUfAhA8MTBPk8bbDabDfo1hWzXv//fyAAAAAAAQIAAAABAAAAAAAAAAAAAAAA\-  \AAAAAAAAAAAAAAAAAAAAAAAAAAD/////DFwBAQgvRUIzMi4wL/////8BAPIFKgEAAAAjIQME7KZA\-  \ozHsyrOOwT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAAAAAAINcGedRly1+dXQrcCaZRXTIG2GHV\-  \0tPCGpZtFnvfhuhSx8d3Azdv/MXRJgsb56qqmD5gsXiWUdi7ia7wsBZVylvWFbNe//9/IAEAAAAB\-  \AgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP////8MXQEBCC9FQjMyLjAv////\-  \/wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf1OHjgkqxnwEYkK9zrAAAAAAAAAAgDxu3\-  \+7op0n6+s1ZJTqqzjHWH84YorH8hTbLiuYGgNyWIkhaj0zR7Vc+fSRm4UYUaPsefRhq3fUt8glyS\-  \D8P/5tcVs17//38gAwAAAAECAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA////\-  \/wxeAQEIL0VCMzIuMC//////AQDyBSoBAAAAIyEDBOymQKMx7MqzjsE+lp+i7WOOwd/U4eOCSrGf\-  \ARiQr3OsAAAAAAAAACDYVJqsPyQ8MwR+LRafufm1LB97SQoyFJdvKVohBvNyfD4/FxT2i0rlYQcS\-  \TQAvTnehousK2P8T9c0qx4Yj72lT1xWzXv//fyAAAAAAAQIAAAABAAAAAAAAAAAAAAAAAAAAAAAA\-  \AAAAAAAAAAAAAAAAAAD/////DF8BAQgvRUIzMi4wL/////8BAPIFKgEAAAAjIQME7KZAozHsyrOO\-  \wT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAA"

@@ -1,1 +0,0 @@-{-# OPTIONS_GHC -F -pgmF hspec-discover #-}

@@ -0,0 +1,391 @@+-- | Helpers shared by the unit and regtest test suites.+module Haskoin.Node.Test+  ( -- * Logging+    withTestLogger,++    -- * Timing+    within,+    eventually,+    receiveWithin,+    expectNothingWithin,++    -- * Block chains+    mkBlocks,+    mkHeaders,+    coinbaseTx,+    serveHeaders,+    headersMessage,+    nowTimestamp,++    -- * Database+    withTestDB,+    withDBAt,++    -- * In-memory connections+    Remote (..),+    memoryConnection,+    remoteSend,+    remoteSendRaw,+    remoteReceive,+    remoteReceiveMatch,+    remoteClose,+    remoteVersion,+    respondHonestly,+    honestRemote,+    encodeMessage,++    -- * Nodes+    TestNode (..),+    withTestNode,+    setTimeout,+    setMaxPeerLife,+    waitForPeer,+    waitForDisconnect,+    waitForBest,+    zeroAddress,+  )+where++import Conduit (awaitForever, yield)+import Control.Monad (forM_, unless)+import Data.ByteString (ByteString)+import Data.ByteString qualified as B+import Data.Default (def)+import Data.List (elemIndex)+import Data.Maybe (fromMaybe, listToMaybe, mapMaybe)+import Data.Serialize (decode, encode, runGet, runPut)+import Data.Time.Clock (NominalDiffTime)+import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Word (Word32, Word64)+import Database.RocksDB (DB)+import Database.RocksDB qualified as R+import Haskoin+import Haskoin.Node+import Haskoin.Node qualified as NC (NodeConfig (..))+import NQE+import Network.Socket (SockAddr (..))+import System.Environment (lookupEnv)+import System.Random (randomIO)+import Test.HUnit (assertFailure)+import UnliftIO+import UnliftIO.Concurrent (threadDelay)++-- | Run with logging to stderr if HASKOIN_TEST_LOG is set, silently otherwise.+withTestLogger :: (Logger -> IO a) -> IO a+withTestLogger f =+  lookupEnv "HASKOIN_TEST_LOG" >>= \case+    Just _ -> withLogger LevelDebug f+    Nothing -> f Logger {level = LevelError, source = "", push = \_ -> return ()}++-- | Fail the test if the action does not complete in the given seconds.+within :: Double -> String -> IO a -> IO a+within s what act =+  timeout (round (s * 1000000)) act >>= \case+    Nothing -> assertFailure $ "Timed out after " <> show s <> "s: " <> what+    Just x -> return x++-- | Fail the test unless the condition becomes true in the given seconds.+eventually :: Double -> String -> IO Bool -> IO ()+eventually s what cond = within s what go+  where+    go = cond >>= \b -> unless b (threadDelay 20000 >> go)++-- | Receive the first matching message in the given seconds.+receiveWithin ::+  (InChan mbox) =>+  Double ->+  String ->+  mbox msg ->+  (msg -> Maybe a) ->+  IO a+receiveWithin s what mbox f = within s what (receiveMatch mbox f)++-- | Fail the test if a matching message arrives in the given seconds.+expectNothingWithin ::+  (InChan mbox) =>+  Double ->+  String ->+  mbox msg ->+  (msg -> Maybe a) ->+  IO ()+expectNothingWithin s what mbox f =+  timeout (round (s * 1000000)) (receiveMatch mbox f) >>= \case+    Nothing -> return ()+    Just _ -> assertFailure $ "Unexpected: " <> what++-- | Current time as a block timestamp.+nowTimestamp :: IO Word32+nowTimestamp = floor <$> getPOSIXTime++-- | Coinbase-like transaction that makes each block unique.+coinbaseTx :: Word32 -> Int -> Tx+coinbaseTx tag i =+  Tx+    { version = 1,+      inputs =+        [ TxIn+            { outpoint = OutPoint {hash = z, index = maxBound},+              script = encode (tag, fromIntegral i :: Word64),+              sequence = maxBound+            }+        ],+      outputs = [TxOut {value = 5000000000, script = B.singleton 0x51}],+      witness = [],+      locktime = 0+    }+  where+    z = "0000000000000000000000000000000000000000000000000000000000000000"++-- | Mine blocks on top of a parent header. Timestamps start at the given+-- time and increase by one second. Use different tags to build forks.+mkBlocks :: Network -> Word32 -> BlockHeader -> Word32 -> Int -> [Block]+mkBlocks net tag parent t0 n = go parent 0+  where+    go p i+      | i >= n = []+      | otherwise =+          let tx = coinbaseTx tag i+              h =+                mineBlock net 0 $+                  BlockHeader+                    { version = 0x20000000,+                      prev = headerHash p,+                      merkle = either error id (buildMerkleRoot [txHash tx]),+                      timestamp = max (t0 + fromIntegral i) (p.timestamp + 1),+                      bits = net.genesisHeader.bits,+                      nonce = 0+                    }+           in Block {header = h, txs = [tx]} : go h (i + 1)++-- | Headers for 'mkBlocks'.+mkHeaders :: Network -> Word32 -> BlockHeader -> Word32 -> Int -> [BlockHeader]+mkHeaders net tag parent t0 n = map (.header) (mkBlocks net tag parent t0 n)++-- | Answer a header request like a full node would. The chain excludes the+-- genesis block.+serveHeaders :: Network -> [BlockHeader] -> GetHeaders -> [BlockHeader]+serveHeaders net chain gh =+  take 2000 (stop_at (drop start chain))+  where+    hashes = headerHash net.genesisHeader : map headerHash chain+    start =+      fromMaybe 0 $+        listToMaybe [i | l <- gh.locator, Just i <- [elemIndex l hashes]]+    z = "0000000000000000000000000000000000000000000000000000000000000000"+    stop_at hs+      | gh.stop == z = hs+      | otherwise = case break ((== gh.stop) . headerHash) hs of+          (a, b : _) -> a <> [b]+          (a, []) -> a++headersMessage :: [BlockHeader] -> Message+headersMessage hs = MHeaders (Headers [(h, VarInt 0) | h <- hs])++-- | Temporary RocksDB database.+withTestDB :: (DB -> IO a) -> IO a+withTestDB f =+  withSystemTempDirectory "haskoin-node-test-" $ \dir -> withDBAt dir f++-- | RocksDB database at a path, created if missing.+withDBAt :: FilePath -> (DB -> IO a) -> IO a+withDBAt dir = R.withDB dir def {R.createIfMissing = True}++-- | Remote end of an in-memory connection to a local peer process.+data Remote = Remote+  { net :: !Network,+    -- | bytes for the local peer, 'Nothing' closes the connection+    outbox :: !(Mailbox (Maybe ByteString)),+    -- | bytes from the local peer, 'Nothing' when it closed the connection+    inbox :: !(Inbox (Maybe ByteString)),+    buffer :: !(IORef ByteString),+    -- | set when the local peer closes the connection+    closed :: !(TVar Bool)+  }++-- | Connection to a remote peer implemented by the provided script. The+-- connection closes when the script returns.+memoryConnection :: Network -> (Remote -> IO ()) -> WithConnection+memoryConnection net script f = do+  (to_local_in, to_local) <- newMailbox+  (from_local_in, from_local) <- newMailbox+  buf <- newIORef B.empty+  closed <- newTVarIO False+  let remote =+        Remote+          { net = net,+            outbox = to_local,+            inbox = from_local_in,+            buffer = buf,+            closed = closed+          }+      src =+        receive to_local_in >>= \case+          Nothing -> return ()+          Just b -> yield b >> src+      snk = awaitForever $ \b -> Just b `send` from_local+      close_local = do+        atomically (writeTVar closed True)+        Nothing `send` from_local+  withAsync (script remote `finally` (Nothing `send` to_local)) $ \_ ->+    f (Conduits src snk) `finally` close_local++encodeMessage :: Network -> Message -> ByteString+encodeMessage net = runPut . putMessage net++remoteSend :: Remote -> Message -> IO ()+remoteSend r m = Just (encodeMessage r.net m) `send` r.outbox++remoteSendRaw :: Remote -> ByteString -> IO ()+remoteSendRaw r bs = Just bs `send` r.outbox++remoteClose :: Remote -> IO ()+remoteClose r = Nothing `send` r.outbox++-- | Receive next message from the local peer. 'Nothing' if it closed the+-- connection.+remoteReceive :: Remote -> IO (Maybe Message)+remoteReceive r =+  take_bytes 24 >>= \case+    Nothing -> return Nothing+    Just hdr -> do+      len <- case decode hdr of+        Left e -> assertFailure $ "Bad header from local peer: " <> e+        Right (MessageHeader _ _ len _) -> return len+      take_bytes (fromIntegral len) >>= \case+        Nothing -> return Nothing+        Just payload ->+          case runGet (getMessage r.net) (hdr <> payload) of+            Left e -> assertFailure $ "Bad message from local peer: " <> e+            Right m -> return (Just m)+  where+    take_bytes n = do+      b <- readIORef r.buffer+      if B.length b >= n+        then do+          let (x, y) = B.splitAt n b+          writeIORef r.buffer y+          return (Just x)+        else+          receive r.inbox >>= \case+            Nothing -> return Nothing+            Just more -> writeIORef r.buffer (b <> more) >> take_bytes n++-- | Receive messages from the local peer until one matches.+remoteReceiveMatch ::+  Double -> String -> Remote -> (Message -> Maybe a) -> IO a+remoteReceiveMatch s what r f = within s what go+  where+    go =+      remoteReceive r >>= \case+        Nothing -> assertFailure $ "Connection closed awaiting " <> what+        Just m -> maybe go return (f m)++zeroAddress :: Word64 -> NetworkAddress+zeroAddress srv = NetworkAddress srv (sockToHostAddress (SockAddrInet 0 0))++-- | Version message for a remote peer.+remoteVersion :: Network -> Word64 -> Word64 -> IO Version+remoteVersion net services nonce = do+  now <- nowTimestamp+  return $+    buildVersion+      net+      nonce+      0+      (zeroAddress services)+      (zeroAddress 0)+      (fromIntegral now)++-- | React to a message like a well-behaved full node with the provided+-- blocks (excluding genesis).+respondHonestly :: Remote -> TVar [Block] -> Message -> IO ()+respondHonestly r blocks = \case+  MVersion _ -> remoteSend r MVerAck+  MPing (Ping n) -> remoteSend r (MPong (Pong n))+  MGetHeaders gh -> do+    bs <- readTVarIO blocks+    remoteSend r . headersMessage $ serveHeaders r.net (map (.header) bs) gh+  MGetData (GetData ivs) -> do+    bs <- readTVarIO blocks+    let find_block (InvVector t h)+          | t `elem` [InvBlock, InvWitnessBlock] =+              listToMaybe [b | b <- bs, (headerHash b.header).get == h]+          | otherwise = Nothing+        found = mapMaybe (\iv -> (,) iv <$> find_block iv) ivs+        missing = [iv | iv <- ivs, iv `notElem` map fst found]+    forM_ found $ \(_, b) -> remoteSend r (MBlock b)+    unless (null missing) $ remoteSend r (MNotFound (NotFound missing))+  _ -> return ()++-- | Remote that completes the handshake and serves the provided blocks.+honestRemote :: TVar [Block] -> Remote -> IO ()+honestRemote blocks r = do+  v <- remoteVersion r.net nodeNetwork =<< randomIO+  remoteSend r (MVersion v)+  let go =+        remoteReceive r >>= \case+          Nothing -> return ()+          Just m -> respondHonestly r blocks m >> go+  go++data TestNode = TestNode+  { node :: !Node,+    events :: !(Inbox NodeEvent),+    net :: !Network+  }++-- | Run a node against the provided peers using a temporary database. The+-- configuration can be adjusted before the node starts.+withTestNode ::+  Network ->+  [String] ->+  (SockAddr -> WithConnection) ->+  (NodeConfig -> NodeConfig) ->+  (TestNode -> IO a) ->+  IO a+withTestNode net peers connect modify f =+  withTestDB $ \db ->+    withTestLogger $ \logger ->+      withPublisher $ \pub ->+        withSubscription pub $ \sub -> do+          let cfg =+                NodeConfig+                  { maxPeers = 20,+                    db = db,+                    cf = Nothing,+                    peers = peers,+                    discover = False,+                    address = zeroAddress 0,+                    net = net,+                    pub = pub,+                    timeout = 120,+                    maxPeerLife = 48 * 3600,+                    connect = connect,+                    logger = logger+                  }+          withNode (modify cfg) $ \n ->+            f TestNode {node = n, events = sub, net = net}++setTimeout :: NominalDiffTime -> NodeConfig -> NodeConfig+setTimeout t c = c {NC.timeout = t}++setMaxPeerLife :: NominalDiffTime -> NodeConfig -> NodeConfig+setMaxPeerLife t c = c {NC.maxPeerLife = t}++waitForPeer :: TestNode -> IO Peer+waitForPeer t =+  receiveWithin 30 "peer connection" t.events $ \case+    PeerEvent (PeerConnected p) -> Just p+    _ -> Nothing++waitForDisconnect :: TestNode -> Peer -> IO ()+waitForDisconnect t p =+  receiveWithin 30 ("disconnection of peer " <> show p) t.events $ \case+    PeerEvent (PeerDisconnected p') | p == p' -> Just ()+    _ -> Nothing++-- | Wait until the best block header satisfies the condition.+waitForBest :: Double -> TestNode -> (BlockNode -> Bool) -> IO BlockNode+waitForBest s t f = do+  eventually s "best block header" $ f <$> chainGetBest t.node.chain+  chainGetBest t.node.chain

@@ -0,0 +1,49 @@+# Regtest image with official Bitcoin Core and Bitcoin Cash Node binaries.+#+# Build with:+#   podman build -t haskoin-node-regtest regtest+#+# Hashes come from the official SHA256SUMS files published by each project:+#   https://bitcoincore.org/bin/bitcoin-core-32.1/SHA256SUMS+#   https://gitlab.com/bitcoin-cash-node/announcements/-/tree/master/release-sigs/29.2.0+FROM docker.io/library/ubuntu:26.04++ARG TARGETARCH+ARG CORE_VERSION=32.1+ARG CORE_SHA256_amd64=61545c17c3120fb7412f896d248b11993af04ec8817aad3649c55b2afb044963+ARG CORE_SHA256_arm64=7ed37c7373eb6aa3e3f72842596a1afeba1b223717033cade9842b25773e1922+ARG BCHN_VERSION=29.2.0+ARG BCHN_SHA256_amd64=23bbe9ff749136a9b832a2a59023ce76a5391322c1ea2f4f24b5b91f035081bc+ARG BCHN_SHA256_arm64=27e177e5a07dc78c2c0d18798ea2e3925111955509ecdc476e828e7ce8f46383++RUN apt-get update \+ && apt-get install -y --no-install-recommends ca-certificates curl libatomic1 \+ && rm -rf /var/lib/apt/lists/*++RUN set -eu; \+    arch="${TARGETARCH:-$(dpkg --print-architecture)}"; \+    case "$arch" in \+      amd64) triple=x86_64-linux-gnu; core_sha="$CORE_SHA256_amd64"; bchn_sha="$BCHN_SHA256_amd64" ;; \+      arm64) triple=aarch64-linux-gnu; core_sha="$CORE_SHA256_arm64"; bchn_sha="$BCHN_SHA256_arm64" ;; \+      *) echo "unsupported architecture: $arch" >&2; exit 1 ;; \+    esac; \+    cd /tmp; \+    core="bitcoin-${CORE_VERSION}-${triple}.tar.gz"; \+    curl -fsSLO "https://bitcoincore.org/bin/bitcoin-core-${CORE_VERSION}/${core}"; \+    echo "${core_sha}  ${core}" | sha256sum -c -; \+    mkdir -p /opt/bitcoin-core; \+    tar -xzf "$core" -C /opt/bitcoin-core --strip-components=1; \+    bchn="bitcoin-cash-node-${BCHN_VERSION}-${triple}.tar.gz"; \+    curl -fsSLO "https://github.com/bitcoin-cash-node/bitcoin-cash-node/releases/download/v${BCHN_VERSION}/${bchn}"; \+    echo "${bchn_sha}  ${bchn}" | sha256sum -c -; \+    mkdir -p /opt/bchn; \+    tar -xzf "$bchn" -C /opt/bchn --strip-components=1; \+    rm -f /tmp/*.tar.gz; \+    /opt/bitcoin-core/bin/bitcoind -version | head -1; \+    /opt/bchn/bin/bitcoind -version | head -1++# Peer-to-peer ports for up to three nodes per container.+EXPOSE 18444 18445 18446++# Nodes are started with "podman exec" by the test suite.+CMD ["sleep", "infinity"]

@@ -0,0 +1,196 @@+-- | Test the library against Bitcoin Core and Bitcoin Cash Node in regtest+-- mode. Nodes run in Podman containers built from the Containerfile in this+-- directory. Set HASKOIN_NODE_SKIP_REGTEST to skip.+module Main (main) where++import Control.Monad (forM_, replicateM_, void)+import Data.List (isInfixOf)+import Data.Maybe (isJust)+import Data.String (fromString)+import Haskoin+import Haskoin.Node+import Haskoin.Node.Test+import Regtest.Podman+import System.Environment (lookupEnv)+import Test.Hspec++main :: IO ()+main = do+  skip <- isJust <$> lookupEnv "HASKOIN_NODE_SKIP_REGTEST"+  available <- podmanAvailable+  if+    | skip -> hspec $ it "regtest" $ pendingWith "HASKOIN_NODE_SKIP_REGTEST set"+    | not available -> hspec $ it "regtest" $ pendingWith "podman not available"+    | otherwise -> buildImage >> hspec spec++-- | Node connected to the given nodes in the container. The timeout is long+-- so that the stall recovery poll does not hide announcement handling bugs.+withRegtestNode :: Container -> [Int] -> (TestNode -> IO a) -> IO a+withRegtestNode c is f = do+  addrs <- mapM (hostAddress c) is+  withTestNode+    (implNet c.impl)+    addrs+    withConnection+    (setTimeout 600)+    f++-- | Wait until our best header is the tip of node number @i@.+waitForTip :: Container -> Int -> TestNode -> IO BlockNode+waitForTip c i t = do+  tip <- fromString <$> cli c i ["getbestblockhash"]+  waitForBest 60 t ((== tip) . headerHash . (.header))++blockHashAt :: Container -> Int -> Int -> IO BlockHash+blockHashAt c i h = fromString <$> cli c i ["getblockhash", show h]++-- | Node 0 with blocks mined so that it leaves initial block download.+withMinedNode :: Impl -> Int -> (Container -> IO a) -> IO a+withMinedNode impl n f =+  withContainer impl $ \c -> do+    startNode c 0 []+    _ <- mine c 0 n+    f c++spec :: Spec+spec = forM_ [BitcoinCore, BitcoinCashNode] $ \impl ->+  describe (implName impl) $ do+    let net = implNet impl++    it "completes the handshake" $ withMinedNode impl 1 $ \c ->+      withRegtestNode c [0] $ \t -> do+        p <- waitForPeer t+        Just o <- getOnlinePeer t.node.peerMgr p+        Just v <- return o.version+        v.version `shouldSatisfy` (>= 70012)+        let VarString ua = v.userAgent+            expected = case impl of+              BitcoinCore -> "Satoshi"+              BitcoinCashNode -> "Bitcoin Cash Node"+        show ua `shouldSatisfy` (expected `isInfixOf`)++    it "syncs a chain longer than one header batch" $+      withMinedNode impl 2100 $ \c ->+        withRegtestNode c [0] $ \t -> do+          bn <- receiveWithin 120 "synced" t.events $ \case+            ChainEvent (ChainSynced bn) -> Just bn+            _ -> Nothing+          bn.height `shouldBe` 2100+          void $ waitForTip c 0 t++    it "follows blocks mined one at a time" $ withMinedNode impl 1 $ \c ->+      withRegtestNode c [0] $ \t -> do+        _ <- waitForTip c 0 t+        replicateM_ 5 $ mine c 0 1 >> waitForTip c 0 t++    it "follows bursts of blocks without stalling" $+      withMinedNode impl 1 $ \c ->+        withRegtestNode c [0] $ \t -> do+          _ <- waitForTip c 0 t+          replicateM_ 3 $ mine c 0 99 >> waitForTip c 0 t++    it "follows a burst longer than a header batch" $+      withMinedNode impl 1 $ \c ->+        withRegtestNode c [0] $ \t -> do+          _ <- waitForTip c 0 t+          _ <- mine c 0 2500+          bn <- waitForTip c 0 t+          bn.height `shouldBe` 2501++    it "follows a reorganization" $ withMinedNode impl 10 $ \c ->+      withRegtestNode c [0] $ \t -> do+        old <- waitForTip c 0 t+        h8 <- cli c 0 ["getblockhash", "8"]+        void $ cli c 0 ["invalidateblock", h8]+        _ <- mineTo c 0 5 otherAddress+        new <- waitForTip c 0 t+        new.height `shouldBe` 12+        chainBlockMain t.node.chain (headerHash old.header) `shouldReturn` False+        split <- chainGetSplitBlock t.node.chain old new+        split.height `shouldBe` 7++    it "follows a reorganization found in a burst" $+      withMinedNode impl 120 $ \c ->+        withRegtestNode c [0] $ \t -> do+          _ <- waitForTip c 0 t+          h <- cli c 0 ["getblockhash", "70"]+          void $ cli c 0 ["invalidateblock", h]+          _ <- mineTo c 0 99 otherAddress+          new <- waitForTip c 0 t+          new.height `shouldBe` 168++    it "downloads blocks" $ withMinedNode impl 5 $ \c ->+      withRegtestNode c [0] $ \t -> do+        p <- waitForPeer t+        _ <- waitForTip c 0 t+        hs <- mapM (blockHashAt c 0) [1 .. 5]+        Just bs <- getBlocks net 10 p hs+        map (headerHash . (.header)) bs `shouldBe` hs+        forM_ bs $ \b ->+          buildMerkleRoot (map txHash b.txs) `shouldBe` Right b.header.merkle+        let unknown = headerHash (mkHeaders net 9 net.genesisHeader 0 1 !! 0)+        getBlocks net 10 p [unknown] `shouldReturn` Nothing++    it "downloads announced transactions" $ withContainer impl $ \c -> do+      startNode c 0 []+      -- Bitcoin Cash Node also loads a default wallet, so name ours+      let wallet args = cli c 0 ("-rpcwallet=w" : args)+      void $ cli c 0 ["createwallet", "w"]+      addr <- wallet ["getnewaddress"]+      void $ mineTo c 0 101 addr+      withRegtestNode c [0] $ \t -> do+        p <- waitForPeer t+        _ <- waitForTip c 0 t+        txid <- fromString <$> wallet ["sendtoaddress", mineAddress, "1"]+        receiveWithin 60 "transaction announcement" t.events $ \case+          PeerEvent (PeerMessage p' (MInv (Inv ivs)))+            | p == p' && any ((== txid) . TxHash . (.invHash)) ivs -> Just ()+          _ -> Nothing+        Just [tx] <- getTxs net 10 p [txid]+        txHash tx `shouldBe` txid++    it "pings a peer" $ withMinedNode impl 1 $ \c ->+      withRegtestNode c [0] $ \t -> do+        p <- waitForPeer t+        pingPeer 10 p `shouldReturn` True++    it "reconnects to a restarted peer and catches up" $+      withMinedNode impl 5 $ \c ->+        withRegtestNode c [0] $ \t -> do+          p <- waitForPeer t+          _ <- waitForTip c 0 t+          stopNode c 0+          waitForDisconnect t p+          startNode c 0 []+          _ <- mine c 0 3+          _ <- waitForPeer t+          bn <- waitForTip c 0 t+          bn.height `shouldBe` 8++    it "keeps following the chain when a peer goes away" $+      withMinedNode impl 5 $ \c -> do+        startNode c 1 ["-addnode=" <> nodeAddress c 0]+        eventually 60 "nodes in sync" $+          (== "5") <$> cli c 1 ["getblockcount"]+        withRegtestNode c [0, 1] $ \t -> do+          _ <- waitForPeer t+          _ <- waitForPeer t+          _ <- waitForTip c 0 t+          stopNode c 0+          _ <- mine c 1 5+          bn <- waitForTip c 1 t+          bn.height `shouldBe` 10++    it "follows bursts announced by several peers" $+      withMinedNode impl 1 $ \c -> do+        startNode c 1 ["-addnode=" <> nodeAddress c 0]+        withRegtestNode c [0, 1] $ \t -> do+          _ <- waitForPeer t+          _ <- waitForPeer t+          _ <- waitForTip c 0 t+          replicateM_ 3 $ mine c 0 99 >> waitForTip c 0 t+          eventually 60 "nodes in sync" $+            (== "298") <$> cli c 1 ["getblockcount"]+          expectNothingWithin 1 "peer disconnection" t.events $ \case+            PeerEvent (PeerDisconnected _) -> Just ()+            _ -> Nothing

@@ -0,0 +1,231 @@+-- | Bitcoin Core and Bitcoin Cash Node regtest nodes in Podman containers.+module Regtest.Podman+  ( Impl (..),+    implName,+    implNet,+    Container (..),+    podmanAvailable,+    buildImage,+    withContainer,+    startNode,+    stopNode,+    cli,+    hostAddress,+    nodeAddress,+    mine,+    mineTo,+    mineAddress,+    otherAddress,+  )+where++import Control.Exception (bracket)+import Control.Monad (unless, void)+import Data.Char (isHexDigit, isSpace)+import Data.List (dropWhileEnd)+import Haskoin (Network, bchRegTest, btcRegTest)+import System.Directory (findExecutable)+import System.Exit (ExitCode (..))+import System.Process (readProcessWithExitCode)+import System.Random (randomRIO)+import UnliftIO (timeout)+import UnliftIO.Concurrent (threadDelay)++data Impl = BitcoinCore | BitcoinCashNode+  deriving stock (Eq, Show)++implName :: Impl -> String+implName BitcoinCore = "Bitcoin Core"+implName BitcoinCashNode = "Bitcoin Cash Node"++implNet :: Impl -> Network+implNet BitcoinCore = btcRegTest+implNet BitcoinCashNode = bchRegTest++implDir :: Impl -> FilePath+implDir BitcoinCore = "/opt/bitcoin-core/bin/"+implDir BitcoinCashNode = "/opt/bchn/bin/"++-- | Running container with ports for up to three nodes.+data Container = Container+  { cid :: !String,+    impl :: !Impl,+    -- | peer-to-peer port of each node inside the container+    p2pPorts :: ![Int],+    -- | RPC port of each node inside the container+    rpcPorts :: ![Int]+  }++-- | Number of nodes that can run in a container.+maxNodes :: Int+maxNodes = 3++image :: String+image = "localhost/haskoin-node-regtest:latest"++-- | Directory with the Containerfile, relative to the package root.+imageDir :: FilePath+imageDir = "test/regtest"++-- | Run podman and return its standard output. Throws on failure or if it+-- takes longer than two minutes.+podman :: [String] -> IO String+podman = podmanWithin 120++-- | Run podman with a timeout in seconds.+podmanWithin :: Int -> [String] -> IO String+podmanWithin secs args =+  timeout (secs * 1000000) (readProcessWithExitCode "podman" args "") >>= \case+    Nothing ->+      fail $+        "podman " <> unwords args <> " timed out after " <> show secs <> "s"+    Just (ExitSuccess, out, _) -> return (dropWhileEnd isSpace out)+    Just (ExitFailure c, out, err) ->+      fail $+        "podman "+          <> unwords args+          <> " exited with "+          <> show c+          <> ": "+          <> out+          <> err++podmanAvailable :: IO Bool+podmanAvailable =+  findExecutable "podman" >>= \case+    Nothing -> return False+    Just _ -> do+      (c, _, _) <- readProcessWithExitCode "podman" ["info"] ""+      return (c == ExitSuccess)++-- | Build the image. Cached layers make this quick after the first time.+buildImage :: IO ()+buildImage = void $ podmanWithin 1800 ["build", "--quiet", "--tag", image, imageDir]++-- | Run a container. Node ports inside it are random, and are published to+-- random ports on the loopback interface of the host.+withContainer :: Impl -> (Container -> IO a) -> IO a+withContainer impl = bracket start stop+  where+    start = do+      ports <- randomPorts (2 * maxNodes)+      let (p2p, rpc) = splitAt maxNodes ports+      cid <-+        podman $+          ["run", "--detach", "--rm"]+            <> concat [["--publish", "127.0.0.1::" <> show p] | p <- p2p]+            <> [image]+      return Container {cid = cid, impl = impl, p2pPorts = p2p, rpcPorts = rpc}+    stop :: Container -> IO ()+    stop c = void $ podman ["rm", "--force", "--time", "0", c.cid]++-- | Distinct random ports. They are at least two apart because nodes may+-- also listen on the next port up for Tor.+randomPorts :: Int -> IO [Int]+randomPorts n = go []+  where+    go ps+      | length ps == n = return (reverse ps)+      | otherwise = do+          p <- randomRIO (20000, 60000)+          if any (\q -> abs (p - q) < 2) ps then go ps else go (p : ps)++p2pPort :: Container -> Int -> Int+p2pPort c i = c.p2pPorts !! i++rpcPort :: Container -> Int -> Int+rpcPort c i = c.rpcPorts !! i++dataDir :: Int -> FilePath+dataDir i = "/data/node" <> show i++exec :: Container -> [String] -> IO String+exec c args = podman (["exec", c.cid] <> args)++-- | Like 'exec' with a timeout in seconds.+execWithin :: Int -> Container -> [String] -> IO String+execWithin secs c args = podmanWithin secs (["exec", c.cid] <> args)++-- | Start node number @i@ in the container with extra arguments. Data+-- persists across restarts.+startNode :: Container -> Int -> [String] -> IO ()+startNode c i extra = do+  void $ exec c ["mkdir", "-p", dataDir i]+  void . exec c $+    [ implDir c.impl <> "bitcoind",+      "-regtest",+      "-daemon",+      "-datadir=" <> dataDir i,+      "-port=" <> show (p2pPort c i),+      "-bind=0.0.0.0:" <> show (p2pPort c i),+      "-rpcport=" <> show (rpcPort c i),+      "-listen=1",+      "-listenonion=0",+      "-fallbackfee=0.0002",+      "-debug=net"+    ]+      <> extra+  void $ cli' 60 c i ["-rpcwait", "getblockcount"]++-- | Stop node number @i@ and wait until it exits.+stopNode :: Container -> Int -> IO ()+stopNode c i = do+  void $ cli c i ["stop"]+  let pidfile = dataDir i <> "/regtest/bitcoind.pid"+      wait n = do+        (code, _, _) <-+          readProcessWithExitCode "podman" ["exec", c.cid, "test", "-e", pidfile] ""+        unless (code /= ExitSuccess || n > (300 :: Int)) $+          threadDelay 100000 >> wait (n + 1)+  wait 0+  threadDelay 500000++cli' :: Int -> Container -> Int -> [String] -> IO String+cli' secs c i args =+  execWithin secs c $+    [ implDir c.impl <> "bitcoin-cli",+      "-regtest",+      "-datadir=" <> dataDir i,+      "-rpcport=" <> show (rpcPort c i)+    ]+      <> args++-- | Call RPC method on node number @i@.+cli :: Container -> Int -> [String] -> IO String+cli = cli' 120++-- | Address of node number @i@ for other nodes in the container.+nodeAddress :: Container -> Int -> String+nodeAddress c i = "127.0.0.1:" <> show (p2pPort c i)++-- | Address and port on the host for node number @i@.+hostAddress :: Container -> Int -> IO String+hostAddress c i = do+  out <- podman ["port", c.cid, show (p2pPort c i) <> "/tcp"]+  case lines out of+    a : _ -> return a+    [] -> fail $ "No port mapping for node " <> show i++-- | Regtest address valid for both implementations.+mineAddress :: String+mineAddress = "mipcBbFg9gMiCh81Kj8tqqdgoZub1ZJRfn"++-- | Another regtest address. Mining a replacement chain to a different+-- address avoids recreating blocks identical to invalidated ones.+otherAddress :: String+otherAddress = "mfcGAzvis9JQAb6avB6WBGiGrgWzLxuGaC"++-- | Mine blocks on node number @i@ and return their hashes.+mine :: Container -> Int -> Int -> IO [String]+mine c i n = mineTo c i n mineAddress++-- | Mine blocks to an address on node number @i@ and return their hashes.+mineTo :: Container -> Int -> Int -> String -> IO [String]+mineTo c i n addr = do+  out <- cli' 600 c i ["generatetoaddress", show n, addr]+  return (filter is_hash (words (map clean out)))+  where+    clean x+      | isHexDigit x = x+      | otherwise = ' '+    is_hash w = length w == 64

@@ -0,0 +1,45 @@+module Haskoin.Node.AddressSpec (spec) where++import Control.Monad (forM_)+import Haskoin (Network (..), btcRegTest)+import Haskoin.Node (toHostService, toSockAddr)+import Network.Socket (SockAddr (..))+import Test.Hspec+import Test.Hspec.QuickCheck (prop)++spec :: Spec+spec = do+  let net = btcRegTest+  describe "addresses" $ do+    it "parses host and service"+      $ forM_+        [ ("localhost", (Just "localhost", Nothing)),+          ("::1", (Just "::1", Nothing)),+          ("localhost:8080", (Just "localhost", Just "8080")),+          ("example.com", (Just "example.com", Nothing)),+          ("api.example.com:443", (Just "api.example.com", Just "443")),+          ("api.example.com:http", (Just "api.example.com", Just "http")),+          ("[::1]", (Just "::1", Nothing)),+          ("[::1]:8080", (Just "::1", Just "8080")),+          ("[2002::dead:beef]:ssh", (Just "2002::dead:beef", Just "ssh"))+        ]+      $ \(s, r) -> toHostService s `shouldBe` r++    it "resolves numeric addresses with the default port" $ do+      as <- toSockAddr net "127.0.0.1"+      map show as `shouldContain` ["127.0.0.1:" <> show net.defaultPort]++    it "returns no addresses for unresolvable hosts" $+      toSockAddr net "host.invalid" `shouldReturn` []++    it "keeps an explicit port" $ do+      as <- toSockAddr net "127.0.0.1:1234"+      map show as `shouldContain` ["127.0.0.1:1234"]++    -- IPv4 only: resolving an IPv6 address needs IPv6 on the host, which+    -- containers and CI runners often lack. The IPv6 forms are still parsed+    -- by the toHostService tests above, which need no resolver.+    prop "resolves any numeric IPv4 address to itself" $ \(e, w) -> do+      let a = SockAddrInet (toEnum (e `mod` 65536)) w+      s <- toSockAddr net (show a)+      s `shouldStartWith` [a]

@@ -0,0 +1,154 @@+module Haskoin.Node.Chain.SyncSpec (spec) where++import Data.List (nub)+import Data.Maybe (isNothing)+import Data.Time.Calendar (fromGregorian)+import Data.Time.Clock (UTCTime (..), addUTCTime)+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)+import Haskoin (BlockHeader (..), BlockNode (..), Network (..), btcRegTest, headerHash)+import Haskoin.Node.Chain.Sync+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++t0 :: UTCTime+t0 = UTCTime (fromGregorian 2026 1 1) 0++at :: Int -> UTCTime+at n = fromIntegral n `addUTCTime` t0++-- | Transitions applied to a few peers.+data Op+  = Connect Int+  | Disconnect Int+  | Request Int+  | Start Int+  | Refresh Int+  | Finish Int+  | Drop Int+  | Poll+  deriving stock (Show)++instance Arbitrary Op where+  arbitrary =+    oneof+      [ Connect <$> peer,+        Disconnect <$> peer,+        Request <$> peer,+        Start <$> peer,+        Refresh <$> peer,+        Finish <$> peer,+        Drop <$> peer,+        pure Poll+      ]+    where+      peer = chooseInt (1, 4)++apply :: SyncState Int -> Op -> SyncState Int+apply s = \case+  Connect p -> peerConnected p s+  Disconnect p -> peerDisconnected p s+  Request p -> requestSync p s+  Start p -> startSync (at 1) p s+  Refresh p -> refreshSync (at 2) p s+  Finish p -> finishSync p s+  Drop p -> dropPeer p s+  Poll -> pollPeers (at 3) s++states :: [Op] -> [SyncState Int]+states = scanl apply (emptySync t0)++run :: [Op] -> SyncState Int+run = foldl' apply (emptySync t0)++statuses :: SyncState Int -> [PeerStatus]+statuses s = [st | p <- [1 .. 4], Just st <- [peerStatus p s]]++isSyncingStatus :: PeerStatus -> Bool+isSyncingStatus = \case+  Syncing _ _ -> True+  _ -> False++spec :: Spec+spec = describe "header sync state" $ do+  it "queues peers in the order they connect" $+    queuedPeers (run [Connect 1, Connect 2, Connect 3]) `shouldBe` [1, 2, 3]++  it "syncs against one peer at a time" $ do+    let s = run [Connect 1, Connect 2, Start 1, Start 2]+    syncingPeer s `shouldBe` Just (1, at 1)+    queuedPeers s `shouldBe` [2]++  it "only starts syncing against queued peers" $+    syncingPeer (run [Connect 1, Start 1, Finish 1, Start 1]) `shouldBe` Nothing++  it "makes a peer idle when it finishes" $+    peerStatus 1 (run [Connect 1, Start 1, Finish 1]) `shouldBe` Just Idle++  it "queues a peer again if it announced blocks while syncing" $ do+    let s = run [Connect 1, Connect 2, Start 1, Request 1, Finish 1]+    queuedPeers s `shouldBe` [2, 1]++  it "queues an idle peer that announces blocks" $+    queuedPeers (run [Connect 1, Start 1, Finish 1, Request 1]) `shouldBe` [1]++  it "queues every idle peer when polling" $+    queuedPeers (run [Connect 1, Connect 2, Start 1, Finish 1, Start 2, Finish 2, Poll])+      `shouldBe` [1, 2]++  it "never queues a dropped peer again" $ do+    let s = run [Connect 1, Drop 1, Request 1, Poll, Connect 1]+    peerStatus 1 s `shouldBe` Just Dropped+    queuedPeers s `shouldBe` []+    connectedPeers s `shouldBe` []++  it "forgets disconnected peers" $+    peerStatus 1 (run [Connect 1, Start 1, Disconnect 1]) `shouldBe` Nothing++  it "records activity when requesting headers" $ do+    lastActivity (run [Connect 1, Start 1]) `shouldBe` at 1+    lastActivity (run [Connect 1, Start 1, Refresh 1]) `shouldBe` at 2+    lastActivity (touchSync (at 9) (emptySync t0) :: SyncState Int) `shouldBe` at 9++  describe "synced" $ do+    let net = btcRegTest+        tipAt t w =+          BlockNode+            { header = net.genesisHeader {timestamp = t},+              height = 1,+              work = w,+              skip = headerHash net.genesisHeader+            }+        now = at 0+        recent = tipAt (floor (utcTimeToPOSIXSeconds now) - 60) 1+        idle = run [Connect 1, Start 1, Finish 1]+    it "is synced with a recent tip, enough work and nothing to do" $+      isSynced net now recent idle `shouldBe` True+    it "is not synced with an old tip" $+      isSynced net (maxTipAge `addUTCTime` now) recent idle `shouldBe` False+    it "is not synced below the minimum chain work" $+      isSynced net {minimumChainWork = 2} now recent idle `shouldBe` False+    it "is not synced while a peer is queued or syncing" $ do+      isSynced net now recent (run [Connect 1]) `shouldBe` False+      isSynced net now recent (run [Connect 1, Start 1]) `shouldBe` False++  prop "never syncs against more than one peer" $ \ops ->+    all ((<= 1) . length . filter isSyncingStatus . statuses) (states ops)++  prop "lists each peer once" $ \ops ->+    all (\s -> let ps = [p | p <- [1 .. 4], Just _ <- [peerStatus p s]] in ps == nub ps) (states ops)++  prop "never queues or syncs a dropped peer until it reconnects" $+    \ops more -> forAll (chooseInt (1, 4)) $ \p ->+      let connected = run (ops <> [Connect p])+          s = foldl' apply (dropPeer p connected) (filter (not . disconnects p) more)+       in peerStatus p s === Just Dropped+            .&&. p `notElem` queuedPeers s+            .&&. fmap fst (syncingPeer s) =/= Just p+            .&&. p `notElem` connectedPeers s++  prop "forgets a peer after it disconnects" $ \ops (p :: Int) ->+    isNothing (peerStatus p (run (ops <> [Disconnect p])))+  where+    disconnects p (Disconnect q) = p == q+    disconnects _ _ = False

@@ -0,0 +1,407 @@+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Header chain process driven by fake peers.+module Haskoin.Node.ChainSpec (spec) where++import Control.Monad (forM_, forever, void)+import Data.Text (Text)+import Data.Time.Clock (NominalDiffTime)+import Database.RocksDB (DB)+import Haskoin+import Haskoin.Node+import Haskoin.Node.Peer.Lock (Lock (..))+import Haskoin.Node.Test+import NQE+import Test.HUnit (assertFailure)+import Test.Hspec hiding (pending)+import UnliftIO++data ChainTest = ChainTest+  { chain :: !Chain,+    events :: !(Inbox ChainEvent),+    peerPub :: !(Publisher PeerEvent),+    net :: !Network+  }++-- | Peer whose outgoing messages are captured instead of sent.+data FakePeer = FakePeer+  { peer :: !Peer,+    inbox :: !(Inbox PeerMessage)+  }++withChainTest :: Network -> NominalDiffTime -> (ChainTest -> IO a) -> IO a+withChainTest net to f = withTestDB $ \db -> withChainOn net to db f++withChainOn ::+  Network -> NominalDiffTime -> DB -> (ChainTest -> IO a) -> IO a+withChainOn net to db f =+  withTestLogger $ \logger ->+    withPublisher $ \pub ->+      withSubscription pub $ \events ->+        withPublisher $ \peerPub ->+          withChain (ChainConfig db Nothing net pub to logger) $ \chain ->+            f ChainTest {chain, events, peerPub, net}++newFakePeer :: ChainTest -> Text -> IO FakePeer+newFakePeer ct label = do+  (inbox, mailbox) <- newMailbox+  busy <- newTVarIO Free+  pending <- newTVarIO 0+  return+    FakePeer+      { peer = Peer {mailbox, pub = ct.peerPub, label, busy, pending},+        inbox+      }++connectPeer :: ChainTest -> Text -> IO FakePeer+connectPeer ct label = do+  fp <- newFakePeer ct label+  chainPeerConnected ct.chain fp.peer+  return fp++describeMessage :: PeerMessage -> String+describeMessage KillPeer = "disconnection"+describeMessage (SendMessage m) = show (commandToString (msgType m))++-- | Next header request sent to the peer, skipping 'MSendHeaders'.+expectGetHeaders :: FakePeer -> IO GetHeaders+expectGetHeaders fp =+  within 10 ("header request to " <> show fp.peer) go+  where+    go =+      receive fp.inbox >>= \case+        SendMessage (MGetHeaders g) -> return g+        SendMessage MSendHeaders -> go+        m -> assertFailure $ "Expected header request, got " <> describeMessage m++expectSendHeaders :: FakePeer -> IO ()+expectSendHeaders fp =+  receiveWithin 10 "sendheaders" fp.inbox $ \case+    SendMessage MSendHeaders -> Just ()+    _ -> Nothing++expectKilled :: FakePeer -> IO ()+expectKilled fp =+  receiveWithin 10 ("disconnection of " <> show fp.peer) fp.inbox $ \case+    KillPeer -> Just ()+    _ -> Nothing++-- | Fail if the peer is asked for headers or disconnected in time window.+expectNoRequest :: Double -> FakePeer -> IO ()+expectNoRequest s fp =+  expectNothingWithin s ("request to " <> show fp.peer) fp.inbox $ \case+    SendMessage (MGetHeaders _) -> Just ()+    KillPeer -> Just ()+    _ -> Nothing++sendHeaders :: ChainTest -> FakePeer -> [BlockHeader] -> IO ()+sendHeaders ct fp = chainHeaders ct.chain fp.peer++-- | Wait for a header request and answer it from the provided chain.+answer :: ChainTest -> FakePeer -> [BlockHeader] -> IO [BlockHeader]+answer ct fp hs = do+  g <- expectGetHeaders fp+  let r = serveHeaders ct.net hs g+  sendHeaders ct fp r+  return r++-- | Answer all header requests from the chain in the variable. Counts the+-- requests.+withServer ::+  ChainTest -> FakePeer -> TVar [BlockHeader] -> (TVar Int -> IO a) -> IO a+withServer ct fp var f = do+  count <- newTVarIO 0+  let go = forever $ do+        g <- receiveMatch fp.inbox $ \case+          SendMessage (MGetHeaders g) -> Just g+          _ -> Nothing+        atomically $ modifyTVar count (+ 1)+        hs <- readTVarIO var+        sendHeaders ct fp (serveHeaders ct.net hs g)+  withAsync go $ \a -> link a >> f count++waitHeight :: ChainTest -> BlockHeight -> IO BlockNode+waitHeight ct h = do+  eventually 30 ("best height " <> show h) $+    (== h) . (.height) <$> chainGetBest ct.chain+  chainGetBest ct.chain++bestHash :: ChainTest -> IO BlockHash+bestHash ct = headerHash . (.header) <$> chainGetBest ct.chain++-- | Sync a fresh peer to the provided chain.+syncPeer :: ChainTest -> Text -> [BlockHeader] -> IO FakePeer+syncPeer ct label hs = do+  fp <- connectPeer ct label+  _ <- answer ct fp hs+  _ <- waitHeight ct (fromIntegral (length hs))+  expectSendHeaders fp+  return fp++invalidPow :: BlockHeader -> BlockHeader+invalidPow h = h {bits = 0x1d00ffff}++spec :: Spec+spec = forM_ [btcRegTest, bchRegTest] $ \net ->+  describe ("header chain on " <> net.name) $ do+    t0 <- runIO $ subtract 6000 <$> nowTimestamp+    let main = mkHeaders net 0 net.genesisHeader t0 4500+        short = take 20 main+        hashAt i = headerHash (main !! (i - 1))+        nodeHash = headerHash . (.header)+        fast = withChainTest net 1+        slow = withChainTest net 600++    describe "initial sync" $ do+      it "requests headers in batches until the peer runs out" $ slow $ \ct -> do+        a <- connectPeer ct "a"+        g1 <- expectGetHeaders a+        take 1 g1.locator `shouldBe` [headerHash net.genesisHeader]+        sendHeaders ct a (serveHeaders net main g1)+        g2 <- expectGetHeaders a+        take 1 g2.locator `shouldBe` [hashAt 2000]+        sendHeaders ct a (serveHeaders net main g2)+        g3 <- expectGetHeaders a+        take 1 g3.locator `shouldBe` [hashAt 4000]+        r3 <- return (serveHeaders net main g3)+        length r3 `shouldBe` 500+        sendHeaders ct a r3+        expectSendHeaders a+        expectNoRequest 0.5 a+        best <- waitHeight ct 4500+        nodeHash best `shouldBe` hashAt 4500++      it "publishes best block and synced events once" $ slow $ \ct -> do+        a <- connectPeer ct "a"+        var <- newTVarIO main+        withServer ct a var $ \count -> do+          bn <- receiveWithin 30 "synced event" ct.events $ \case+            ChainSynced bn -> Just bn+            _ -> Nothing+          bn.height `shouldBe` 4500+          chainIsSynced ct.chain `shouldReturn` True+          readTVarIO count `shouldReturn` 3+          expectNothingWithin 0.5 "second synced event" ct.events $ \case+            ChainSynced _ -> Just ()+            _ -> Nothing++      it "does not consider old headers synced" $ slow $ \ct -> do+        let old = mkHeaders net 0 net.genesisHeader 1500000000 10+        _ <- syncPeer ct "a" old+        chainIsSynced ct.chain `shouldReturn` False++      it "stops asking a peer that sends no headers" $ slow $ \ct -> do+        a <- connectPeer ct "a"+        _ <- expectGetHeaders a+        sendHeaders ct a []+        expectSendHeaders a+        expectNoRequest 1 a++      it "syncs against every peer that connects" $ slow $ \ct -> do+        _ <- syncPeer ct "a" (take 10 main)+        b <- connectPeer ct "b"+        _ <- answer ct b (take 30 main)+        void $ waitHeight ct 30++      it "ignores duplicate connection events" $ slow $ \ct -> do+        a <- connectPeer ct "a"+        chainPeerConnected ct.chain a.peer+        _ <- answer ct a short+        expectSendHeaders a+        expectNoRequest 1 a++      it "keeps headers across restarts" $ withTestDB $ \db -> do+        withChainOn net 600 db $ \ct -> void $ syncPeer ct "a" short+        withChainOn net 600 db $ \ct -> do+          bn <- receiveWithin 10 "best block" ct.events $ \case+            ChainBestBlock bn -> Just bn+            _ -> Nothing+          bn.height `shouldBe` 20+          bestHash ct `shouldReturn` hashAt 20++    describe "announcements" $ do+      it "requests headers when an unknown block is announced by inventory" $+        slow $ \ct -> do+          a <- syncPeer ct "a" (take 10 main)+          chainBlocksAnnounced ct.chain a.peer [hashAt 109]+          _ <- answer ct a (take 109 main)+          void $ waitHeight ct 109++      it "ignores inventory announcements of known blocks" $ slow $ \ct -> do+        a <- syncPeer ct "a" (take 10 main)+        chainBlocksAnnounced ct.chain a.peer [hashAt 5, hashAt 10]+        expectNoRequest 1 a++      it "imports announced headers that connect" $ slow $ \ct -> do+        a <- syncPeer ct "a" (take 10 main)+        sendHeaders ct a [main !! 10]+        _ <- waitHeight ct 11+        expectNoRequest 0.5 a++      it "requests headers when announced headers do not connect" $+        slow $ \ct -> do+          a <- syncPeer ct "a" (take 10 main)+          sendHeaders ct a (take 5 (drop 49 main))+          _ <- answer ct a (take 54 main)+          void $ waitHeight ct 54++      it "continues syncing a peer that sends a full batch unsolicited" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          b <- connectPeer ct "b"+          _ <- expectGetHeaders a+          sendHeaders ct b (take 2000 main)+          _ <- waitHeight ct 2000+          sendHeaders ct a []+          g <- expectGetHeaders b+          take 1 g.locator `shouldBe` [hashAt 2000]++      it "keeps syncing peer when another peer announces headers" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          b <- connectPeer ct "b"+          g <- expectGetHeaders a+          sendHeaders ct b (take 5 main)+          _ <- waitHeight ct 5+          sendHeaders ct a (serveHeaders net short g)+          _ <- waitHeight ct 20+          expectSendHeaders a+          -- b gets its turn once a is done+          void $ answer ct b (take 5 main)++      it "ignores messages from peers that are not connected" $+        slow $ \ct -> do+          x <- newFakePeer ct "x"+          sendHeaders ct x (take 5 main)+          chainBlocksAnnounced ct.chain x.peer [hashAt 5]+          expectNoRequest 1 x+          (.height) <$> chainGetBest ct.chain `shouldReturn` 0++    describe "reorganizations" $ do+      let fork n = mkHeaders net 1 (main !! 14) (t0 + 100) n+      it "switches to a fork with more work" $ slow $ \ct -> do+        a <- syncPeer ct "a" short+        sendHeaders ct a (fork 10)+        best <- waitHeight ct 25+        nodeHash best `shouldBe` headerHash (last (fork 10))+        chainBlockMain ct.chain (hashAt 20) `shouldReturn` False+        chainBlockMain ct.chain (hashAt 15) `shouldReturn` True+        Just old <- chainGetBlock ct.chain (hashAt 20)+        split <- chainGetSplitBlock ct.chain old best+        nodeHash split `shouldBe` hashAt 15++      it "stays on the chain with more work" $ slow $ \ct -> do+        a <- syncPeer ct "a" short+        sendHeaders ct a (fork 3)+        let tip = headerHash (last (fork 3))+        eventually 10 "fork stored" $+          (/= Nothing) <$> chainGetBlock ct.chain tip+        bestHash ct `shouldReturn` hashAt 20+        chainBlockMain ct.chain tip `shouldReturn` False++    describe "misbehaving peers" $ do+      it "disconnects a peer that sends headers with invalid proof of work" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          _ <- expectGetHeaders a+          sendHeaders ct a [invalidPow (main !! 0)]+          expectKilled a++      it "disconnects a peer that sends headers that do not form a chain" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          _ <- expectGetHeaders a+          sendHeaders ct a [main !! 0, main !! 2]+          expectKilled a++      it "syncs from another peer after disconnecting a misbehaving one" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          b <- connectPeer ct "b"+          _ <- expectGetHeaders a+          sendHeaders ct a [invalidPow (main !! 0)]+          expectKilled a+          _ <- answer ct b short+          void $ waitHeight ct 20++      it "disconnects an unresponsive syncing peer and moves on" $+        fast $ \ct -> do+          a <- connectPeer ct "a"+          b <- connectPeer ct "b"+          _ <- expectGetHeaders a+          expectKilled a+          _ <- answer ct b short+          void $ waitHeight ct 20++      it "moves on when the syncing peer disconnects" $ slow $ \ct -> do+        a <- connectPeer ct "a"+        b <- connectPeer ct "b"+        _ <- expectGetHeaders a+        chainPeerDisconnected ct.chain a.peer+        _ <- answer ct b short+        void $ waitHeight ct 20++      it "never asks a peer for headers after it disconnects" $+        slow $ \ct -> do+          a <- connectPeer ct "a"+          b <- connectPeer ct "b"+          _ <- expectGetHeaders a+          -- queue a again while it is syncing+          chainBlocksAnnounced ct.chain a.peer [hashAt 5]+          chainPeerDisconnected ct.chain a.peer+          _ <- answer ct b short+          _ <- waitHeight ct 20+          expectNoRequest 1 a++      it "waits for a peer reserved by someone else" $ fast $ \ct -> do+        other <- newOwner "other"+        a <- newFakePeer ct "a"+        True <- reservePeer other a.peer+        chainPeerConnected ct.chain a.peer+        expectNoRequest 1 a+        True <- releasePeer other a.peer+        void $ answer ct a short++      it "keeps its reservation when others try to release the peer" $+        slow $ \ct -> do+          other <- newOwner "other"+          a <- connectPeer ct "a"+          _ <- expectGetHeaders a+          False <- releasePeer other a.peer+          setFree a.peer+          fmap ownerName <$> peerHolder a.peer `shouldReturn` Just "Chain"+          False <- reservePeer other a.peer+          sendHeaders ct a []+          eventually 10 "peer released" $ (== Nothing) . fmap ownerName <$> peerHolder a.peer++    describe "stall recovery" $ do+      it "asks peers for headers when nothing happens for a while" $+        fast $ \ct -> do+          a <- syncPeer ct "a" (take 10 main)+          -- the peer finds blocks but never announces them+          var <- newTVarIO (take 30 main)+          withServer ct a var $ \_ -> void (waitHeight ct 30)++      it "recovers when the syncing peer stops responding" $ fast $ \ct -> do+        a <- connectPeer ct "a"+        _ <- expectGetHeaders a+        expectKilled a+        chainPeerDisconnected ct.chain a.peer+        b <- connectPeer ct "b"+        void $ answer ct b short++    describe "queries" $ do+      it "answers questions about the chain" $ slow $ \ct -> do+        _ <- syncPeer ct "a" short+        tip <- chainGetBest ct.chain+        Just anc <- chainGetAncestor ct.chain 10 tip+        nodeHash anc `shouldBe` hashAt 10+        ps <- chainGetParents ct.chain 15 tip+        map (.height) ps `shouldBe` [15 .. 19]+        map nodeHash ps `shouldBe` map hashAt [15 .. 19]+        chainGetBlock ct.chain (headerHash (main !! 100)) `shouldReturn` Nothing+        chainBlockMain ct.chain (headerHash (main !! 100)) `shouldReturn` False+        chainBlockMain ct.chain (hashAt 20) `shouldReturn` True+        Just n5 <- chainGetBlock ct.chain (hashAt 5)+        split <- chainGetSplitBlock ct.chain n5 tip+        nodeHash split `shouldBe` hashAt 5

@@ -0,0 +1,60 @@+module Haskoin.Node.Peer.LockSpec (spec) where++import Haskoin.Node.Peer.Lock+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++-- | An attempt to reserve or release, by one of a few owners.+data Op = Acquire Int | Release Int+  deriving stock (Show)++instance Arbitrary Op where+  arbitrary = oneof [Acquire <$> owner, Release <$> owner]+    where+      owner = chooseInt (1, 3)++-- | Apply an operation, keeping the lock unchanged when it does not apply.+apply :: Lock Int -> Op -> Lock Int+apply l = \case+  Acquire o -> maybe l id (acquire o l)+  Release o -> maybe l id (release o l)++spec :: Spec+spec = describe "peer lock" $ do+  it "lets an owner reserve a free peer" $+    acquire 1 Free `shouldBe` Just (HeldBy (1 :: Int))++  it "refuses to reserve a held peer, even for its holder" $ do+    acquire 2 (HeldBy (1 :: Int)) `shouldBe` Nothing+    acquire 1 (HeldBy (1 :: Int)) `shouldBe` Nothing++  it "lets the holder release the peer" $+    release 1 (HeldBy (1 :: Int)) `shouldBe` Just Free++  it "refuses to release for anyone but the holder" $ do+    release 2 (HeldBy (1 :: Int)) `shouldBe` Nothing+    release 1 (Free :: Lock Int) `shouldBe` Nothing++  prop "only the last owner to reserve can release" $ \(ops :: [Op]) ->+    let final = foldl' apply Free ops+        model = foldl' step Nothing ops+        step Nothing (Acquire o) = Just o+        step (Just h) (Release o) | h == o = Nothing+        step m _ = m+     in holder final === model++  describe "owners" $ do+    it "are all different" $ do+      a <- newOwner "a"+      b <- newOwner "a"+      a `shouldNotBe` b+      a `shouldBe` a++    it "share the legacy owner" $+      legacyOwner `shouldBe` legacyOwner++    it "keep their names" $ do+      a <- newOwner "chain"+      ownerName a `shouldBe` "chain"+      ownerName legacyOwner `shouldBe` "legacy"

@@ -0,0 +1,90 @@+module Haskoin.Node.Peer.RequestSpec (spec) where++import Haskoin+import Haskoin.Node.Peer.Request+import Haskoin.Node.Test (mkBlocks)+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++blocks :: [Block]+blocks = mkBlocks btcRegTest 0 btcRegTest.genesisHeader 1700000000 5++blockInv :: Block -> InvVector+blockInv b = InvVector InvWitnessBlock (headerHash b.header).get++txInv :: Tx -> InvVector+txInv tx = InvVector InvWitnessTx (txHash tx).get++-- | Feed messages to a request for the given items.+feed :: [InvVector] -> [Message] -> RequestStep+feed ivs = foldl' step (newRequest 42 ivs)+  where+    step (Continue r) m = stepRequest r m+    step done _ = done++-- | Messages that have nothing to do with the request.+noise :: Gen Message+noise =+  elements+    [ MPing (Ping 1),+      MPong (Pong 1),+      MVerAck,+      MSendHeaders,+      MInv (Inv [blockInv (blocks !! 4)]),+      MNotFound (NotFound [txInv (coinbaseOf (blocks !! 4))])+    ]++coinbaseOf :: Block -> Tx+coinbaseOf b = case b.txs of+  tx : _ -> tx+  [] -> error "block without transactions"++spec :: Spec+spec = describe "data request" $ do+  let want = take 3 blocks+      ivs = map blockInv want+      done = Complete (map Right want)++  it "completes when all items arrive in order" $+    feed ivs (map MBlock want) `shouldBe` done++  it "completes immediately when nothing is requested" $+    newRequest 42 [] `shouldBe` Complete []++  it "receives transactions" $ do+    let txs = map coinbaseOf want+    feed (map txInv txs) (map MTx txs) `shouldBe` Complete (map Left txs)++  it "waits for missing items" $+    case feed ivs (map MBlock (take 2 want)) of+      Continue _ -> return ()+      other -> expectationFailure ("Expected to continue, got " <> show other)++  it "fails when items arrive out of order" $+    feed ivs [MBlock (want !! 1)] `shouldBe` Failed "block out of order"++  it "fails when a requested item is not found" $+    feed ivs [MBlock (head' want), MNotFound (NotFound [ivs !! 2])]+      `shouldBe` Failed "item not found"++  it "fails when the pong arrives before all items" $+    feed ivs [MBlock (head' want), MPong (Pong 42)]+      `shouldBe` Failed "incomplete response"++  it "ignores a pong for another ping" $+    feed ivs (MPong (Pong 7) : map MBlock want) `shouldBe` done++  it "does not accept a transaction for a block request" $+    case feed ivs [MTx (coinbaseOf (head' want))] of+      Continue _ -> return ()+      other -> expectationFailure ("Expected to continue, got " <> show other)++  prop "ignores unrelated messages between items" $+    forAll (vectorOf 3 (listOf noise)) $ \gaps ->+      let msgs = concat (zipWith (\gap b -> gap <> [MBlock b]) gaps want)+       in feed ivs msgs === done+  where+    head' xs = case xs of+      x : _ -> x+      [] -> error "empty"

@@ -0,0 +1,139 @@+module Haskoin.Node.PeerMgr.AddressesSpec (spec) where++import Data.Maybe (isJust, isNothing)+import Data.Time.Calendar (fromGregorian)+import Data.Time.Clock (NominalDiffTime, UTCTime (..), addUTCTime)+import Haskoin.Node.PeerMgr.Addresses+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++t0 :: UTCTime+t0 = UTCTime (fromGregorian 2026 1 1) 0++later :: NominalDiffTime -> UTCTime+later d = d `addUTCTime` t0++book :: [Int] -> AddressBook Int+book = foldl' (flip (addAddress Discovered)) emptyBook++-- | Take an address that must be available.+taken :: UTCTime -> AddressBook Int -> (Int, AddressBook Int)+taken now b = case takeAddress now b of+  Just x -> x+  Nothing -> error "no address to take"++-- | Take an address, end its connection, and return the new book.+cycleWith :: Outcome -> Int -> AddressBook Int -> AddressBook Int+cycleWith outcome a b = case takeAddress t0 b of+  Just (a', b') | a == a' -> finishAddress t0 outcome a b'+  _ -> error "address not taken"++-- | When an address can be tried again, if it is waiting.+retryAt :: Int -> AddressBook Int -> Maybe UTCTime+retryAt a b = case addressState a b of+  Just (Waiting t _) -> Just t+  _ -> Nothing++-- | Transitions applied to a few addresses at increasing times.+data Op+  = Add Source Int+  | Take NominalDiffTime+  | Give Int+  | Connected Int+  | Finish NominalDiffTime Outcome Int+  deriving stock (Show)++instance Arbitrary Op where+  arbitrary =+    oneof+      [ Add <$> elements [Configured, Discovered] <*> addr,+        Take <$> time,+        Give <$> addr,+        Connected <$> addr,+        Finish <$> time <*> elements [Disconnected, Failed, Misbehaved] <*> addr+      ]+    where+      addr = chooseInt (1, 4)+      time = fromIntegral <$> chooseInt (0, 100000)++apply :: AddressBook Int -> Op -> AddressBook Int+apply b = \case+  Add src a -> addAddress src a b+  Take d -> maybe b snd (takeAddress (later d) b)+  Give a -> returnAddress a b+  Connected a -> markConnected a b+  Finish d o a -> finishAddress (later d) o a b++spec :: Spec+spec = describe "address book" $ do+  it "hands out each address once" $ do+    let (a, b) = taken t0 (book [1])+    a `shouldBe` 1+    addressState 1 b `shouldBe` Just InUse+    takeAddress t0 b `shouldSatisfy` isNothing++  it "takes back an address that was not used" $ do+    let (_, b) = taken t0 (book [1])+    addressState 1 (returnAddress 1 b) `shouldBe` Just Ready++  it "waits briefly later a normal disconnection" $ do+    let b = cycleWith Disconnected 1 (book [1])+    retryAt 1 b `shouldBe` Just (later retryDelay)+    takeAddress t0 b `shouldSatisfy` isNothing+    takeAddress (later retryDelay) b `shouldSatisfy` isJust++  it "doubles the wait later each failure, up to a limit" $ do+    let waits = map (retryAt 1) (drop 1 (iterate (cycleFailed 1) (book [1])))+        cycleFailed a b = finishAddress t0 Failed a (snd (taken (later 1000000) b))+    take 4 waits `shouldBe` map (Just . later) [2, 4, 8, 16]+    waits !! 20 `shouldBe` Just (later maxBackoff)++  it "forgets failures once the peer comes online" $ do+    let failed = cycleWith Failed 1 (book [1])+        (_, b) = taken (later 1000) failed+        b' = finishAddress t0 Failed 1 (markConnected 1 b)+    retryAt 1 b' `shouldBe` Just (later 2)++  it "bans a discovered address that misbehaved" $+    retryAt 1 (cycleWith Misbehaved 1 (book [1])) `shouldBe` Just (later banTime)++  it "never bans a configured address" $ do+    let b = addAddress Configured 1 (book [1])+    retryAt 1 (cycleWith Misbehaved 1 b) `shouldBe` Just (later maxBackoff)++  it "keeps a bounded number of discovered addresses" $ do+    let b = book [1 .. maxDiscovered + 10]+        b' = addAddress Configured (maxDiscovered + 20) b+    length (knownAddresses b) `shouldBe` maxDiscovered+    length (knownAddresses b') `shouldBe` maxDiscovered + 1++  describe "loading" $ do+    it "loads when the book is empty" $+      needsReload t0 (emptyBook :: AddressBook Int) `shouldBe` True++    it "does not load while an address can be tried" $+      needsReload t0 (book [1]) `shouldBe` False++    it "does not load again too soon" $ do+      -- banned, so that nothing can be tried for a long time+      let b = markLoaded t0 (cycleWith Misbehaved 1 (book [1]))+      needsReload (later 1) b `shouldBe` False+      needsReload (later reloadInterval) b `shouldBe` True++  prop "only hands out addresses that can be tried" $ \(ops :: [Op]) (d :: Int) ->+    let b = foldl' apply emptyBook ops+        now = later (fromIntegral (abs d))+     in case takeAddress now b of+          Nothing -> property True+          Just (a, b') ->+            let ok = case addressState a b of+                  Just Ready -> True+                  Just (Waiting t _) -> t <= now+                  _ -> False+             in ok .&&. addressState a b' === Just InUse++  prop "never forgets an address" $ \(ops :: [Op]) ->+    let b = foldl' apply emptyBook ops+        added = [a | Add _ a <- ops]+     in all (\a -> isJust (addressState a b)) added

@@ -0,0 +1,95 @@+module Haskoin.Node.PeerMgr.HandshakeSpec (spec) where++import Control.Monad (foldM)+import Data.Word (Word64)+import Haskoin+import Haskoin.Node (buildVersion, myVersion)+import Haskoin.Node.PeerMgr.Handshake+import Haskoin.Node.Test (zeroAddress)+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++-- | Version of a full node with nonce 7.+version :: Version+version = versionWith nodeNetwork++versionWith :: Word64 -> Version+versionWith services =+  buildVersion btcRegTest 7 0 (zeroAddress services) (zeroAddress 0) 0++-- | Run events from the start of the handshake, collecting actions.+run :: [HandshakeEvent] -> Either HandshakeError (Handshake, [HandshakeAction])+run = foldM step (AwaitingBoth, [])+  where+    step (h, acc) e = fmap (acc <>) <$> stepHandshake h e++-- | Random handshake messages: 'True' is a version, 'False' a verack.+toEvents :: [Bool] -> [HandshakeEvent]+toEvents = map (\b -> if b then GotVersion version else GotVerAck)++-- | At most one version and at most one verack, in any order: every sequence+-- a well-behaved peer can send.+validEvents :: Gen [Bool]+validEvents = do+  msgs <- sublistOf [True, False]+  elements [msgs, reverse msgs]++count :: (Eq a) => a -> [a] -> Int+count x = length . filter (== x)++spec :: Spec+spec = describe "version handshake" $ do+  it "completes when the version comes first" $+    run [GotVersion version, GotVerAck]+      `shouldBe` Right (Established version, [SendVerAck, AnnounceConnected])++  it "completes when the verack comes first" $+    run [GotVerAck, GotVersion version]+      `shouldBe` Right (Established version, [SendVerAck, AnnounceConnected])++  it "waits for the verack after the version" $+    run [GotVersion version] `shouldBe` Right (AwaitingVerAck version, [SendVerAck])++  it "rejects a repeated verack" $ do+    run [GotVerAck, GotVerAck] `shouldBe` Left DuplicateVerAck+    run [GotVersion version, GotVerAck, GotVerAck]+      `shouldBe` Left DuplicateVerAck++  it "rejects a repeated version" $ do+    run [GotVersion version, GotVersion version] `shouldBe` Left DuplicateVersion+    run [GotVersion version, GotVerAck, GotVersion version]+      `shouldBe` Left DuplicateVersion++  prop "fails exactly when a message is repeated" $ \bs ->+    let repeated = count True bs > 1 || count False bs > 1+     in either (const True) (const False) (run (toEvents bs)) === repeated++  prop "acknowledges the version once" $+    forAll validEvents $ \bs ->+      fmap (count SendVerAck . snd) (run (toEvents bs)) === Right (count True bs)++  prop "announces the peer once, when both messages arrived" $+    forAll validEvents $ \bs ->+      let both = length bs == 2+       in fmap (\(h, as) -> (isEstablished h, count AnnounceConnected as)) (run (toEvents bs))+            === Right (both, if both then 1 else 0)++  prop "knows the version of the peer once received" $+    forAll validEvents $ \bs ->+      fmap (handshakeVersion . fst) (run (toEvents bs))+        === Right (if True `elem` bs then Just version else Nothing)++  prop "knows whether the peer acknowledged our version" $+    forAll validEvents $ \bs ->+      fmap (hasVerAck . fst) (run (toEvents bs)) === Right (False `elem` bs)++  describe "version checks" $ do+    it "accepts a full node" $+      checkVersion [1, 2] version `shouldBe` Right version+    it "rejects a peer that does not serve blocks" $+      checkVersion [] (versionWith 0) `shouldBe` Left NotFullNode+    it "rejects a connection to ourselves" $+      checkVersion [7] version `shouldBe` Left ConnectedToSelf+    it "uses our protocol version" $+      version.version `shouldBe` fromIntegral myVersion

@@ -0,0 +1,157 @@+module Haskoin.Node.PeerMgr.LifecycleSpec (spec) where++import Data.Time.Calendar (fromGregorian)+import Data.Time.Clock (UTCTime (..), addUTCTime)+import Haskoin (Version, btcRegTest, nodeNetwork)+import Haskoin.Node (buildVersion)+import Haskoin.Node.PeerMgr.Handshake+import Haskoin.Node.PeerMgr.Lifecycle+import Haskoin.Node.Test (zeroAddress)+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck++t0 :: UTCTime+t0 = UTCTime (fromGregorian 2026 1 1) 0++-- | Seconds after the connection opened.+at :: Double -> UTCTime+at s = realToFrac s `addUTCTime` t0++version :: Version+version = buildVersion btcRegTest 7 0 (zeroAddress nodeNetwork) (zeroAddress 0) 0++-- | A ten second timeout and a hundred second maximum life.+limits :: Limits+limits = Limits {timeout = 10, maxLife = 100}++-- | Feed inputs to a new connection, collecting the actions.+run :: [PeerInput] -> (PeerState, [PeerAction])+run = feed (newPeerState t0)++-- | Feed inputs to a connection, collecting the actions.+feed :: PeerState -> [PeerInput] -> (PeerState, [PeerAction])+feed st0 = foldl' go (st0, [])+  where+    go (st, acc) i = let (st', as) = stepPeer limits i st in (st', acc <> as)++-- | A connection that completed the handshake.+online :: [PeerInput]+online = [GotHandshake (GotVersion version), GotHandshake GotVerAck]++-- | A random input.+newtype Input = Input PeerInput+  deriving stock (Show)++instance Arbitrary Input where+  arbitrary =+    Input+      <$> oneof+        [ Heard <$> time,+          GotHandshake <$> elements [GotVersion version, GotVerAck],+          Refused <$> elements [NotFullNode, ConnectedToSelf],+          GotPong <$> time <*> nonce,+          Tick <$> time <*> nonce+        ]+    where+      time = at <$> choose (0, 200)+      nonce = chooseBoundedIntegral (1, 3)++spec :: Spec+spec = describe "connection lifecycle" $ do+  describe "handshake" $ do+    it "goes online and announces the peer" $ do+      let (st, actions) = run online+      isOnline st `shouldBe` True+      st.announced `shouldBe` True+      stageVersion st `shouldBe` Just version+      actions `shouldBe` [ReplyVerAck, Announce]++    it "accepts the verack first" $+      snd (run [GotHandshake GotVerAck, GotHandshake (GotVersion version)])+        `shouldBe` [ReplyVerAck, Announce]++    it "disconnects a peer that repeats a handshake message once online" $+      snd (run (online <> [GotHandshake GotVerAck]))+        `shouldBe` [ReplyVerAck, Announce, Disconnect (Rejected DuplicateVerAck)]++    it "disconnects a peer whose version was refused" $ do+      let (st, actions) = run [Refused ConnectedToSelf]+      actions `shouldBe` [Disconnect (Rejected ConnectedToSelf)]+      st.stage `shouldBe` Disconnecting (Rejected ConnectedToSelf)++    it "disconnects a peer that does not complete the handshake in time" $+      -- chatter keeps the peer from being silent, but not from timing out+      snd (run [GotHandshake GotVerAck, Heard (at 9), Tick (at 11) 1])+        `shouldBe` [Disconnect HandshakeTimeout]++  describe "pings" $ do+    it "pings an online peer on a tick" $ do+      let (st, actions) = run (online <> [Tick (at 1) 5])+      drop 2 actions `shouldBe` [SendPing 5]+      pendingPing st `shouldBe` Just (at 1, 5)++    it "pings once until the pong arrives" $+      drop 2 (snd (run (online <> [Tick (at 1) 5, Tick (at 2) 6])))+        `shouldBe` [SendPing 5]++    it "records the round trip of a matching pong" $ do+      let (st, _) = run (online <> [Tick (at 1) 5, GotPong (at 1.5) 5])+      pendingPing st `shouldBe` Nothing+      st.pings `shouldBe` [0.5]++    it "ignores a pong with another nonce" $ do+      let (st, _) = run (online <> [Tick (at 1) 5, GotPong (at 1.5) 6])+      pendingPing st `shouldBe` Just (at 1, 5)+      st.pings `shouldBe` []++    it "keeps the last eleven round trips" $ do+      -- the pong also counts as hearing from the peer+      let ping i = [Tick (at i) 1, GotPong (at (i + i / 100)) 1, Heard (at (i + i / 100))]+          (st, _) = run (online <> concatMap ping [1 .. 15])+      -- in milliseconds, so that rounding in the times does not matter+      map (\d -> round (d * 1000) :: Int) st.pings `shouldBe` [150, 140 .. 50]++    it "disconnects a peer that does not answer a ping in time" $+      -- other traffic does not make up for a missing pong+      drop 3 (snd (run (online <> [Tick (at 1) 5, Heard (at 10), Tick (at 12) 6])))+        `shouldBe` [Disconnect NoPong]++  describe "timers" $ do+    it "disconnects a silent peer" $+      drop 2 (snd (run (online <> [Tick (at 11) 1])))+        `shouldBe` [Disconnect Silent]++    it "rotates a peer connected for too long" $+      drop 2 (snd (run (online <> [Heard (at 100), Tick (at 101) 1])))+        `shouldBe` [Disconnect TooOld]++    it "rotates before checking anything else" $+      snd (run [Tick (at 101) 1]) `shouldBe` [Disconnect TooOld]++  describe "properties" $ do+    prop "ignores everything once disconnecting" $ \(ins :: [Input]) ->+      let inputs = [i | Input i <- ins]+       in let (st, _) = run [Refused NotFullNode]+           in snd (feed st inputs) === []++    prop "announces a peer at most once, and remembers it" $ \(ins :: [Input]) ->+      let inputs = [i | Input i <- ins]+       in let (st, actions) = run inputs+              announces = length (filter (== Announce) actions)+           in announces <= 1 .&&. st.announced === (announces == 1)++    prop "disconnects at most once" $ \(ins :: [Input]) ->+      let inputs = [i | Input i <- ins]+       in let isDisconnect = \case+                Disconnect _ -> True+                _ -> False+           in length (filter isDisconnect (snd (run inputs))) <= 1++    prop "is online exactly when announced and not disconnecting" $ \(ins :: [Input]) ->+      let inputs = [i | Input i <- ins]+       in let (st, _) = run inputs+              disconnecting = case st.stage of+                Disconnecting _ -> True+                _ -> False+           in isOnline st === (st.announced && not disconnecting)

@@ -0,0 +1,228 @@+-- | Full node connected to scripted remote peers over in-memory connections.+module Haskoin.Node.PeerMgrSpec (spec) where++import Control.Monad (forever, void, when)+import Data.Time.Clock (diffUTCTime, getCurrentTime)+import Haskoin+import Haskoin.Node+import Haskoin.Node.Test+import Network.Socket (SockAddr (..))+import System.Random (randomIO)+import Test.Hspec+import UnliftIO+import UnliftIO.Concurrent (threadDelay)++-- | Node connected to a single remote running the given script. Each+-- connection attempt runs the script again with the attempt number.+withScriptedNode ::+  Network ->+  (NodeConfig -> NodeConfig) ->+  (Int -> Remote -> IO ()) ->+  (TestNode -> IO a) ->+  IO a+withScriptedNode net modify script f = do+  attempts <- newTVarIO (0 :: Int)+  let connect :: SockAddr -> WithConnection+      connect _ g = do+        n <- atomically $ stateTVar attempts (\n -> (n + 1, n + 1))+        memoryConnection net (script n) g+  withTestNode net ["127.0.0.1:1"] connect modify f++-- | Remote that sends the version and then runs the handler on each+-- message.+handshakeThen :: Version -> (Message -> IO ()) -> Remote -> IO ()+handshakeThen v react r = do+  remoteSend r (MVersion v)+  let go =+        remoteReceive r >>= \case+          Nothing -> return ()+          Just m -> react m >> go+  go++noPeerConnectedWithin :: Double -> TestNode -> IO ()+noPeerConnectedWithin s t =+  expectNothingWithin s "peer connection" t.events $ \case+    PeerEvent (PeerConnected _) -> Just ()+    _ -> Nothing++spec :: Spec+spec = do+  let net = bchRegTest+  t0 <- runIO $ subtract 3000 <$> nowTimestamp+  let main = mkBlocks net 0 net.genesisHeader t0 2500++  describe "node handshake" $ do+    it "connects to a full node" $ do+      blocks <- newTVarIO []+      withScriptedNode net id (const (honestRemote blocks)) $ \t -> do+        p <- waitForPeer t+        Just o <- getOnlinePeer t.node.peerMgr p+        fmap (.version) o.version `shouldBe` Just (fromIntegral myVersion)+        o.online `shouldBe` True+        map (.mailbox) <$> getPeers t.node.peerMgr `shouldReturn` [p]++    it "rejects a peer that is not a full node" $ do+      v <- remoteVersion net 0 =<< randomIO+      withScriptedNode net id (\_ -> handshakeThen v (const (return ()))) $+        noPeerConnectedWithin 2++    it "rejects a connection to itself" $ do+      -- reply with the nonce from our own version message+      let script _ r =+            remoteReceive r >>= \case+              Just (MVersion ours) -> do+                v <- remoteVersion net nodeNetwork ours.nonce+                remoteSend r (MVersion v)+                remoteSend r MVerAck+                void (remoteReceive r)+              _ -> return ()+      withScriptedNode net id script $ noPeerConnectedWithin 2++    it "disconnects a peer that sends its version twice" $ do+      v <- remoteVersion net nodeNetwork =<< randomIO+      let script n r = handshakeThen v (react n r) r+          react n r = \case+            MVersion _ -> do+              remoteSend r MVerAck+              when (n == 1) $ remoteSend r (MVersion v)+            _ -> return ()+      withScriptedNode net id script $ \t -> do+        p <- waitForPeer t+        waitForDisconnect t p++    it "disconnects a peer that does not complete the handshake" $ do+      remotes <- newEmptyMVar+      let script _ r = do+            putMVar remotes r+            void (remoteReceive r) -- our version+            void (remoteReceive r) -- wait until closed+      withScriptedNode net (setTimeout 1) script $ \t -> do+        r <- within 10 "connection" (readMVar remotes)+        eventually 10 "connection closed" (readTVarIO r.closed)+        noPeerConnectedWithin 0.5 t++    it "disconnects a peer that chatters without completing the handshake" $ do+      remotes <- newEmptyMVar+      let script _ r = do+            void (tryPutMVar remotes r)+            -- never sends a version, but keeps talking+            forever $ remoteSend r (MPing (Ping 1)) >> threadDelay 200000+      withScriptedNode net (setTimeout 1) script $ \t -> do+        r <- within 10 "connection" (readMVar remotes)+        eventually 10 "connection closed" (readTVarIO r.closed)+        noPeerConnectedWithin 0.5 t++    it "waits before retrying a peer that failed" $ do+      times <- newTVarIO []+      let script _ _ = getCurrentTime >>= \now -> atomically (modifyTVar' times (now :))+      withScriptedNode net id script $ \_ -> do+        eventually 20 "two attempts" $ (>= 2) . length <$> readTVarIO times+        -- the first failure waits two seconds; the connect loop alone could+        -- retry within a tenth of a second+        oldestFirst <- reverse <$> readTVarIO times+        case oldestFirst of+          first : second : _ ->+            (second `diffUTCTime` first) `shouldSatisfy` (>= 1.9)+          _ -> expectationFailure "fewer than two attempts"++  describe "node peer maintenance" $ do+    it "answers pings" $ do+      blocks <- newTVarIO []+      pongs <- newEmptyMVar+      let script _ r = do+            v <- remoteVersion net nodeNetwork =<< randomIO+            remoteSend r (MVersion v)+            remoteSend r (MPing (Ping 1234))+            let go =+                  remoteReceive r >>= \case+                    Nothing -> return ()+                    Just (MPong (Pong n)) -> putMVar pongs n >> go+                    Just m -> respondHonestly r blocks m >> go+            go+      withScriptedNode net id script $ \_ ->+        within 10 "pong" (takeMVar pongs) `shouldReturn` 1234++    it "disconnects a peer that stops responding" $ do+      v <- remoteVersion net nodeNetwork =<< randomIO+      let react r = \case+            MVersion _ -> remoteSend r MVerAck+            _ -> return () -- never answers pings+          script _ r = handshakeThen v (react r) r+      withScriptedNode net (setTimeout 2) script $ \t -> do+        p <- waitForPeer t+        waitForDisconnect t p++    it "disconnects a peer that has been connected for too long" $ do+      blocks <- newTVarIO []+      withScriptedNode+        net+        (setMaxPeerLife 1 . setTimeout 2)+        (const (honestRemote blocks))+        $ \t -> do+          p <- waitForPeer t+          waitForDisconnect t p++    it "reconnects after a peer disconnects" $ do+      blocks <- newTVarIO []+      let script n r+            | n == 1 = do+                v <- remoteVersion net nodeNetwork =<< randomIO+                remoteSend r (MVersion v)+                Just (MVersion _) <- remoteReceive r+                remoteSend r MVerAck+                Just MVerAck <- remoteReceive r+                return () -- close the connection+            | otherwise = honestRemote blocks r+      withScriptedNode net id script $ \t -> do+        p1 <- waitForPeer t+        waitForDisconnect t p1+        p2 <- waitForPeer t+        p2 `shouldNotBe` p1++  describe "node header sync" $ do+    it "syncs headers from a peer" $ do+      blocks <- newTVarIO main+      withScriptedNode net id (const (honestRemote blocks)) $ \t -> do+        bn <- receiveWithin 30 "synced" t.events $ \case+          ChainEvent (ChainSynced bn) -> Just bn+          _ -> Nothing+        headerHash bn.header `shouldBe` headerHash (last main).header++    it "follows blocks announced by inventory" $ do+      blocks <- newTVarIO (take 10 main)+      remotes <- newEmptyMVar+      let script _ r = putMVar remotes r >> honestRemote blocks r+      withScriptedNode net id script $ \t -> do+        r <- within 10 "remote" (readMVar remotes)+        _ <- waitForBest 30 t ((== 10) . (.height))+        atomically $ writeTVar blocks (take 109 main)+        let tip = headerHash (main !! 108).header+        remoteSend r (MInv (Inv [InvVector InvBlock tip.get]))+        void $ waitForBest 30 t ((== 109) . (.height))++    it "follows blocks announced by headers that do not connect" $ do+      blocks <- newTVarIO (take 10 main)+      remotes <- newEmptyMVar+      let script _ r = putMVar remotes r >> honestRemote blocks r+      withScriptedNode net id script $ \t -> do+        r <- within 10 "remote" (readMVar remotes)+        p <- waitForPeer t+        _ <- waitForBest 30 t ((== 10) . (.height))+        atomically $ writeTVar blocks (take 60 main)+        remoteSend r (headersMessage [(main !! 59).header])+        _ <- waitForBest 30 t ((== 60) . (.height))+        expectNothingWithin 0.5 "disconnection" t.events $ \case+          PeerEvent (PeerDisconnected p') | p == p' -> Just ()+          _ -> Nothing++    it "downloads blocks from a peer" $ do+      blocks <- newTVarIO main+      withScriptedNode net id (const (honestRemote blocks)) $ \t -> do+        p <- waitForPeer t+        let want = take 3 (drop 100 main)+        getBlocks net 10 p (map (headerHash . (.header)) want)+          `shouldReturn` Just want+        let unknown = case mkBlocks net 7 net.genesisHeader t0 1 of+              b : _ -> headerHash b.header+              [] -> error "no block"+        getBlocks net 10 p [unknown] `shouldReturn` Nothing