-- | Dial logic for the Switch.
--
-- Implements connection reuse, exponential backoff, dial deduplication,
-- and parallel staggered dialing (Happy Eyeballs, RFC 8305).
--
-- Dial flow:
--   1. Check connection pool for existing Open connection
--   2. Check per-peer backoff (reject if recently failed)
--   3. Deduplication: join pending dial if another thread is already dialing
--   4. Select transport per address, staggered parallel dial
--   5. Upgrade first successful raw connection
--   6. Add to pool / record backoff on failure
module LibP2P.Switch.Dial
  ( -- * Main entry point
    dial
    -- * Dial options
  , DialOpts (..)
  , defaultDialOpts
  , dialWith
    -- * Backoff management
  , checkBackoff
  , recordBackoff
  , clearBackoff
    -- * Constants (exported for testing)
  , initialBackoffSeconds
  , maxBackoffSeconds
  , staggerDelayUs
  ) where

import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (Async, async, cancel, waitAnyCatch, waitCatch)
import Control.Concurrent.STM
  ( STM
  , TMVar
  , TVar
  , atomically
  , newEmptyTMVar
  , putTMVar
  , readTMVar
  , readTVar
  , tryPutTMVar
  , writeTChan
  , writeTVar
  )
import Control.Exception (SomeException, bracketOnError, catch, finally, onException)
import Control.Monad (forM, forM_, when)
import Data.List (find)
import qualified Data.Map.Strict as Map
import Data.Time.Clock (NominalDiffTime, addUTCTime, getCurrentTime)
import LibP2P.Crypto.PeerId (PeerId)
import LibP2P.Multiaddr (Multiaddr (..), isRelayedAddr)
import LibP2P.Multiaddr.Protocol (Protocol (..))
import LibP2P.Switch.ConnPool (addConn, lookupConn)
import LibP2P.Switch.Connection (closeConnection)
import LibP2P.Switch.Listen (streamAcceptLoop, switchListenAddrs)
import LibP2P.Switch.ResourceManager (Direction (..), releaseConnection, reserveConnection)
import LibP2P.Switch.Types
  ( BackoffEntry (..)
  , Connection (..)
  , DialError (..)
  , MuxerSession (..)
  , Switch (..)
  , SwitchEvent (..)
  )
import LibP2P.Switch.Upgrade (upgradeAs)
import LibP2P.Transport (RawConnection (..), Transport (..))

-- | Initial backoff duration after first failure: 5 seconds.
initialBackoffSeconds :: NominalDiffTime
initialBackoffSeconds :: NominalDiffTime
initialBackoffSeconds = NominalDiffTime
5

-- | Maximum backoff duration: 300 seconds (5 minutes).
maxBackoffSeconds :: NominalDiffTime
maxBackoffSeconds :: NominalDiffTime
maxBackoffSeconds = NominalDiffTime
300

-- | Stagger delay between parallel dial attempts: 250ms (RFC 8305).
staggerDelayUs :: Int
staggerDelayUs :: Int
staggerDelayUs = Int
250000

-- | Check if a peer is currently in dial backoff.
-- Returns Right () if no backoff is active or the backoff has expired.
-- Expired entries are cleaned up atomically.
checkBackoff :: TVar (Map.Map PeerId BackoffEntry) -> PeerId -> IO (Either DialError ())
checkBackoff :: TVar (Map PeerId BackoffEntry)
-> PeerId -> IO (Either DialError ())
checkBackoff TVar (Map PeerId BackoffEntry)
backoffsVar PeerId
pid = do
  now <- IO UTCTime
getCurrentTime
  atomically $ do
    boffs <- readTVar backoffsVar
    case Map.lookup pid boffs of
      Maybe BackoffEntry
Nothing -> Either DialError () -> STM (Either DialError ())
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either DialError ()
forall a b. b -> Either a b
Right ())
      Just BackoffEntry
be
        | BackoffEntry -> UTCTime
beExpiry BackoffEntry
be UTCTime -> UTCTime -> Bool
forall a. Ord a => a -> a -> Bool
<= UTCTime
now -> do
            -- Expired, clean up
            TVar (Map PeerId BackoffEntry) -> Map PeerId BackoffEntry -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar (Map PeerId BackoffEntry)
backoffsVar (PeerId -> Map PeerId BackoffEntry -> Map PeerId BackoffEntry
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
pid Map PeerId BackoffEntry
boffs)
            Either DialError () -> STM (Either DialError ())
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either DialError ()
forall a b. b -> Either a b
Right ())
        | Bool
otherwise -> Either DialError () -> STM (Either DialError ())
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError ()
forall a b. a -> Either a b
Left DialError
DialBackoff)

-- | Record a backoff after a failed dial.
-- First failure: 5s. Each subsequent: duration * 2, capped at 300s.
-- Backoff formula: min(initialBackoff * 2^(attempts-1), maxBackoff)
recordBackoff :: TVar (Map.Map PeerId BackoffEntry) -> PeerId -> IO ()
recordBackoff :: TVar (Map PeerId BackoffEntry) -> PeerId -> IO ()
recordBackoff TVar (Map PeerId BackoffEntry)
backoffsVar PeerId
pid = do
  now <- IO UTCTime
getCurrentTime
  atomically $ do
    boffs <- readTVar backoffsVar
    let attempts = case PeerId -> Map PeerId BackoffEntry -> Maybe BackoffEntry
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid Map PeerId BackoffEntry
boffs of
          Maybe BackoffEntry
Nothing -> Int
1
          Just BackoffEntry
be -> BackoffEntry -> Int
beAttempts BackoffEntry
be Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1
        -- Exponential backoff: 5s, 10s, 20s, 40s, ..., capped at 300s
        duration = NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Ord a => a -> a -> a
min NominalDiffTime
maxBackoffSeconds
                       (NominalDiffTime
initialBackoffSeconds NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Num a => a -> a -> a
* Int -> NominalDiffTime
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
2 Int -> Int -> Int
forall a b. (Num a, Integral b) => a -> b -> a
^ (Int
attempts Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1) :: Int))
        entry = BackoffEntry
          { beExpiry :: UTCTime
beExpiry   = NominalDiffTime -> UTCTime -> UTCTime
addUTCTime NominalDiffTime
duration UTCTime
now
          , beAttempts :: Int
beAttempts = Int
attempts
          }
    writeTVar backoffsVar (Map.insert pid entry boffs)

-- | Clear backoff for a peer (called on successful connection).
clearBackoff :: TVar (Map.Map PeerId BackoffEntry) -> PeerId -> IO ()
clearBackoff :: TVar (Map PeerId BackoffEntry) -> PeerId -> IO ()
clearBackoff TVar (Map PeerId BackoffEntry)
backoffsVar PeerId
pid = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  boffs <- TVar (Map PeerId BackoffEntry) -> STM (Map PeerId BackoffEntry)
forall a. TVar a -> STM a
readTVar TVar (Map PeerId BackoffEntry)
backoffsVar
  writeTVar backoffsVar (Map.delete pid boffs)

-- | Result of checking for a pending dial (used internally).
data PendingCheck
  = JoinExisting !(TMVar (Either DialError Connection))
  | StartNew     !(TMVar (Either DialError Connection))

-- | Per-dial options.
--
-- Mirrors the two orthogonal context values go-libp2p threads through a
-- dial: @network.WithForceDirectDial@ and @network.WithSimultaneousConnect@,
-- which its hole puncher sets together.
data DialOpts = DialOpts
  { DialOpts -> Bool
doForceDirect :: !Bool
    -- ^ Bypass connection reuse, dial backoff and dial deduplication,
    -- and always establish a new transport connection. Hole punching
    -- needs this: reusing a pooled connection emits no packet at all, so
    -- the TCP simultaneous connect the DCUtR spec relies on cannot
    -- happen. go-libp2p likewise consults backoff only when the dial is
    -- not force-direct.
  , DialOpts -> Bool
doUpgradeAsClient :: !Bool
    -- ^ Whether to run the client side of the security handshake and the
    -- muxer. False upgrades as the responder over a connection we
    -- dialled, which specs/relay/DCUtR requires of peer @B@: "For the
    -- purpose of all protocols run on top of this TCP connection, @A@ is
    -- assumed to be the client and @B@ the server."
  }

-- | Ordinary dial: reuse pooled connections, honour backoff, act as client.
defaultDialOpts :: DialOpts
defaultDialOpts :: DialOpts
defaultDialOpts = DialOpts
  { doForceDirect :: Bool
doForceDirect     = Bool
False
  , doUpgradeAsClient :: Bool
doUpgradeAsClient = Bool
True
  }

-- | Dial a peer, reusing existing connections or establishing new ones.
--
-- Implements the full dial flow:
--   1. Pool reuse: return existing Open connection if available
--   2. Backoff check: reject if peer recently failed
--   3. Deduplication: coalesce concurrent dials to same peer via TMVar
--   4. Staggered parallel dial with 250ms delay (Happy Eyeballs)
--   5. First success: upgrade, add to pool, return
--   6. All fail: record backoff, return error
dial :: Switch -> PeerId -> [Multiaddr] -> IO (Either DialError Connection)
dial :: Switch -> PeerId -> [Multiaddr] -> IO (Either DialError Connection)
dial Switch
sw = Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> IO (Either DialError Connection)
dialWith Switch
sw DialOpts
defaultDialOpts

-- | Dial a peer under explicit options.
--
-- A force-direct dial skips steps 1-3 entirely. Skipping deduplication
-- is required, not incidental: DCUtR calls its dialer once per address
-- so that every address is attempted at the same moment, and a shared
-- pending-dial TMVar carries one result for all waiters, so joining it
-- would collapse those attempts into a single address. Backoff is still
-- *recorded* on failure, as go-libp2p does.
dialWith :: Switch -> DialOpts -> PeerId -> [Multiaddr] -> IO (Either DialError Connection)
dialWith :: Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> IO (Either DialError Connection)
dialWith Switch
sw DialOpts
opts PeerId
remotePeerId [Multiaddr]
addrs = do
  -- 0. Check switch is open
  closed <- STM Bool -> IO Bool
forall a. STM a -> IO a
atomically (STM Bool -> IO Bool) -> STM Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar (Switch -> TVar Bool
swClosed Switch
sw)
  if closed
    then pure (Left DialSwitchClosed)
    else if doForceDirect opts
      then establishAndRegister sw opts remotePeerId addrs
      else do
        -- 1. Check connection pool for existing Open connection
        existing <- atomically $ lookupConn (swConnPool sw) remotePeerId
        case existing of
          Just Connection
conn -> Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Connection -> Either DialError Connection
forall a b. b -> Either a b
Right Connection
conn)
          Maybe Connection
Nothing -> do
            -- 2. Check backoff
            backoffResult <- TVar (Map PeerId BackoffEntry)
-> PeerId -> IO (Either DialError ())
checkBackoff (Switch -> TVar (Map PeerId BackoffEntry)
swDialBackoffs Switch
sw) PeerId
remotePeerId
            case backoffResult of
              Left DialError
err -> Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left DialError
err)
              Right () -> do
                -- 3. Deduplication: check for pending dial
                joinOrCreate <- STM PendingCheck -> IO PendingCheck
forall a. STM a -> IO a
atomically (STM PendingCheck -> IO PendingCheck)
-> STM PendingCheck -> IO PendingCheck
forall a b. (a -> b) -> a -> b
$ Switch -> PeerId -> STM PendingCheck
checkPendingDial Switch
sw PeerId
remotePeerId
                case joinOrCreate of
                  JoinExisting TMVar (Either DialError Connection)
tmvar ->
                    -- Another thread is already dialing; wait for its result
                    STM (Either DialError Connection)
-> IO (Either DialError Connection)
forall a. STM a -> IO a
atomically (STM (Either DialError Connection)
 -> IO (Either DialError Connection))
-> STM (Either DialError Connection)
-> IO (Either DialError Connection)
forall a b. (a -> b) -> a -> b
$ TMVar (Either DialError Connection)
-> STM (Either DialError Connection)
forall a. TMVar a -> STM a
readTMVar TMVar (Either DialError Connection)
tmvar
                  StartNew TMVar (Either DialError Connection)
tmvar ->
                    -- We own this dial; execute and broadcast result.
                    -- If the dial throws, fill the TMVar and drop the
                    -- pending entry so waiters and future dials never
                    -- wedge on a stale pending dial.
                    Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> TMVar (Either DialError Connection)
-> IO (Either DialError Connection)
dialNewAndBroadcast Switch
sw DialOpts
opts PeerId
remotePeerId [Multiaddr]
addrs TMVar (Either DialError Connection)
tmvar
                      IO (Either DialError Connection)
-> IO () -> IO (Either DialError Connection)
forall a b. IO a -> IO b -> IO a
`onException` Switch -> PeerId -> TMVar (Either DialError Connection) -> IO ()
abortPendingDial Switch
sw PeerId
remotePeerId TMVar (Either DialError Connection)
tmvar

-- | Clean up a pending dial whose worker threw an exception.
-- Fills the TMVar (if still empty) so joined waiters are released,
-- and removes the pending map entry so future dials can proceed.
abortPendingDial :: Switch -> PeerId -> TMVar (Either DialError Connection) -> IO ()
abortPendingDial :: Switch -> PeerId -> TMVar (Either DialError Connection) -> IO ()
abortPendingDial Switch
sw PeerId
pid TMVar (Either DialError Connection)
tmvar = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  _ <- TMVar (Either DialError Connection)
-> Either DialError Connection -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar (Either DialError Connection)
tmvar (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left ([String] -> DialError
DialAllFailed [String
"dial aborted by exception"]))
  pending <- readTVar (swPendingDials sw)
  writeTVar (swPendingDials sw) (Map.delete pid pending)

-- | Atomically check for an existing pending dial or create one.
checkPendingDial :: Switch -> PeerId -> STM PendingCheck
checkPendingDial :: Switch -> PeerId -> STM PendingCheck
checkPendingDial Switch
sw PeerId
pid = do
  pending <- TVar (Map PeerId (TMVar (Either DialError Connection)))
-> STM (Map PeerId (TMVar (Either DialError Connection)))
forall a. TVar a -> STM a
readTVar (Switch -> TVar (Map PeerId (TMVar (Either DialError Connection)))
swPendingDials Switch
sw)
  case Map.lookup pid pending of
    Just TMVar (Either DialError Connection)
tmvar -> PendingCheck -> STM PendingCheck
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TMVar (Either DialError Connection) -> PendingCheck
JoinExisting TMVar (Either DialError Connection)
tmvar)
    Maybe (TMVar (Either DialError Connection))
Nothing -> do
      tmvar <- STM (TMVar (Either DialError Connection))
forall a. STM (TMVar a)
newEmptyTMVar
      writeTVar (swPendingDials sw) (Map.insert pid tmvar pending)
      pure (StartNew tmvar)

-- | Execute the dial, broadcast the result to joined waiters, and drop
-- the pending-dial entry.
dialNewAndBroadcast
  :: Switch -> DialOpts -> PeerId -> [Multiaddr]
  -> TMVar (Either DialError Connection)
  -> IO (Either DialError Connection)
dialNewAndBroadcast :: Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> TMVar (Either DialError Connection)
-> IO (Either DialError Connection)
dialNewAndBroadcast Switch
sw DialOpts
opts PeerId
remotePeerId [Multiaddr]
addrs TMVar (Either DialError Connection)
tmvar = do
  result <- Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> IO (Either DialError Connection)
establishAndRegister Switch
sw DialOpts
opts PeerId
remotePeerId [Multiaddr]
addrs
  atomically $ do
    putTMVar tmvar result
    pending <- readTVar (swPendingDials sw)
    writeTVar (swPendingDials sw) (Map.delete remotePeerId pending)
  pure result

-- | Reserve resources, dial, verify the peer id, and register the
-- resulting connection.
--
-- Shared by the ordinary dial path and the force-direct one, which
-- reaches it without touching the pool, backoff or pending-dial state.
--
-- The direction is taken from 'doUpgradeAsClient' and used for the
-- resource reservation, the upgrade roles and 'connDirection' alike, so
-- the release in 'closeConnection' -- which reads 'connDirection' --
-- always matches what was reserved.
establishAndRegister
  :: Switch -> DialOpts -> PeerId -> [Multiaddr]
  -> IO (Either DialError Connection)
establishAndRegister :: Switch
-> DialOpts
-> PeerId
-> [Multiaddr]
-> IO (Either DialError Connection)
establishAndRegister Switch
sw DialOpts
opts PeerId
remotePeerId [Multiaddr]
addrs = do
  let dir :: Direction
dir = if DialOpts -> Bool
doUpgradeAsClient DialOpts
opts then Direction
Outbound else Direction
Inbound
  resCheck <- STM (Either ResourceError ()) -> IO (Either ResourceError ())
forall a. STM a -> IO a
atomically (STM (Either ResourceError ()) -> IO (Either ResourceError ()))
-> STM (Either ResourceError ()) -> IO (Either ResourceError ())
forall a b. (a -> b) -> a -> b
$ ResourceManager
-> PeerId -> Direction -> STM (Either ResourceError ())
reserveConnection (Switch -> ResourceManager
swResourceMgr Switch
sw) PeerId
remotePeerId Direction
dir
  case resCheck of
    Left ResourceError
resErr -> Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left (ResourceError -> DialError
DialResourceLimit ResourceError
resErr))
    Right () -> do
      result <- Switch
-> DialOpts
-> Direction
-> [Multiaddr]
-> IO (Either DialError Connection)
dialNewInner Switch
sw DialOpts
opts Direction
dir [Multiaddr]
addrs
        IO (Either DialError Connection)
-> IO () -> IO (Either DialError Connection)
forall a b. IO a -> IO b -> IO a
`onException` STM () -> IO ()
forall a. STM a -> IO a
atomically (ResourceManager -> PeerId -> Direction -> STM ()
releaseConnection (Switch -> ResourceManager
swResourceMgr Switch
sw) PeerId
remotePeerId Direction
dir)
      -- Verify remote PeerId matches expected target
      let verified = case Either DialError Connection
result of
            Right Connection
conn
              | Connection -> PeerId
connPeerId Connection
conn PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerId
remotePeerId ->
                  DialError -> Either DialError Connection
forall a b. a -> Either a b
Left (PeerId -> PeerId -> DialError
DialPeerIdMismatch PeerId
remotePeerId (Connection -> PeerId
connPeerId Connection
conn))
            Either DialError Connection
_ -> Either DialError Connection
result
      case verified of
        Right Connection
conn -> do
          TVar (Map PeerId BackoffEntry) -> PeerId -> IO ()
clearBackoff (Switch -> TVar (Map PeerId BackoffEntry)
swDialBackoffs Switch
sw) PeerId
remotePeerId
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
            TVar (Map PeerId [Connection]) -> Connection -> STM ()
addConn (Switch -> TVar (Map PeerId [Connection])
swConnPool Switch
sw) Connection
conn
            TChan SwitchEvent -> SwitchEvent -> STM ()
forall a. TChan a -> a -> STM ()
writeTChan (Switch -> TChan SwitchEvent
swEvents Switch
sw)
              (PeerId -> Direction -> Multiaddr -> SwitchEvent
Connected (Connection -> PeerId
connPeerId Connection
conn) Direction
dir (Connection -> Multiaddr
connRemoteAddr Connection
conn))
          -- Start accepting inbound streams on the dialer side; tear the
          -- connection down when the session dies (pool removal,
          -- resource release, muxer + transport close).
          _ <- IO () -> IO (Async ())
forall a. IO a -> IO (Async a)
async (IO () -> IO (Async ())) -> IO () -> IO (Async ())
forall a b. (a -> b) -> a -> b
$ Switch -> Connection -> IO ()
streamAcceptLoop Switch
sw Connection
conn IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`finally` Switch -> Connection -> IO ()
closeConnection Switch
sw Connection
conn
          -- Notify connection listeners (e.g. GossipSub auto-stream open)
          notifiers <- atomically $ readTVar (swNotifiers sw)
          mapM_ (\Connection -> IO ()
f -> IO () -> IO (Async ())
forall a. IO a -> IO (Async a)
async (IO () -> IO (Async ())) -> IO () -> IO (Async ())
forall a b. (a -> b) -> a -> b
$ Connection -> IO ()
f Connection
conn) notifiers
          pure (Right conn)
        Left DialError
_ -> do
          -- Close the muxer session on PeerId mismatch
          case Either DialError Connection
result of
            Right Connection
conn -> MuxerSession -> IO ()
muxClose (Connection -> MuxerSession
connSession Connection
conn)
            Left DialError
_     -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          -- Release the reserved connection since dial failed
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ ResourceManager -> PeerId -> Direction -> STM ()
releaseConnection (Switch -> ResourceManager
swResourceMgr Switch
sw) PeerId
remotePeerId Direction
dir
          TVar (Map PeerId BackoffEntry) -> PeerId -> IO ()
recordBackoff (Switch -> TVar (Map PeerId BackoffEntry)
swDialBackoffs Switch
sw) PeerId
remotePeerId
          Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either DialError Connection
verified

-- | Inner dial logic: transport selection and staggered parallel dial.
dialNewInner :: Switch -> DialOpts -> Direction -> [Multiaddr] -> IO (Either DialError Connection)
dialNewInner :: Switch
-> DialOpts
-> Direction
-> [Multiaddr]
-> IO (Either DialError Connection)
dialNewInner Switch
_sw DialOpts
_opts Direction
_dir [] = Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left DialError
DialNoAddresses)
dialNewInner Switch
sw DialOpts
opts Direction
dir [Multiaddr]
addrs = do
  transports <- STM [Transport] -> IO [Transport]
forall a. STM a -> IO a
atomically (STM [Transport] -> IO [Transport])
-> STM [Transport] -> IO [Transport]
forall a b. (a -> b) -> a -> b
$ TVar [Transport] -> STM [Transport]
forall a. TVar a -> STM a
readTVar (Switch -> TVar [Transport]
swTransports Switch
sw)
  -- Find a transport for each address
  let dialable = (Multiaddr -> Maybe (Multiaddr, Transport))
-> [Multiaddr] -> [(Multiaddr, Transport)]
forall a b. (a -> Maybe b) -> [a] -> [b]
filterMap (\Multiaddr
addr ->
        case (Transport -> Bool) -> [Transport] -> Maybe Transport
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Maybe a
find (\Transport
t -> Transport -> Multiaddr -> Bool
transportCanDial Transport
t Multiaddr
addr) [Transport]
transports of
          Just Transport
t  -> (Multiaddr, Transport) -> Maybe (Multiaddr, Transport)
forall a. a -> Maybe a
Just (Multiaddr
addr, Transport
t)
          Maybe Transport
Nothing -> Maybe (Multiaddr, Transport)
forall a. Maybe a
Nothing) [Multiaddr]
addrs
  case dialable of
    []    -> Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left (Multiaddr -> DialError
DialNoTransport ([Multiaddr] -> Multiaddr
forall a. HasCallStack => [a] -> a
Prelude.head [Multiaddr]
addrs)))
    [(Multiaddr, Transport)]
pairs -> Switch
-> DialOpts
-> Direction
-> [(Multiaddr, Transport)]
-> IO (Either DialError Connection)
staggeredDial Switch
sw DialOpts
opts Direction
dir [(Multiaddr, Transport)]
pairs

-- | Filter and map a list, keeping only Just results.
filterMap :: (a -> Maybe b) -> [a] -> [b]
filterMap :: forall a b. (a -> Maybe b) -> [a] -> [b]
filterMap a -> Maybe b
_ [] = []
filterMap a -> Maybe b
f (a
x:[a]
xs) = case a -> Maybe b
f a
x of
  Maybe b
Nothing -> (a -> Maybe b) -> [a] -> [b]
forall a b. (a -> Maybe b) -> [a] -> [b]
filterMap a -> Maybe b
f [a]
xs
  Just b
y  -> b
y b -> [b] -> [b]
forall a. a -> [a] -> [a]
: (a -> Maybe b) -> [a] -> [b]
forall a b. (a -> Maybe b) -> [a] -> [b]
filterMap a -> Maybe b
f [a]
xs

-- | Attempt to dial using staggered parallel attempts (Happy Eyeballs, RFC 8305).
--
-- Addresses are tried with 250ms delay between each attempt.
-- The first successful connection wins; remaining attempts are cancelled.
staggeredDial
  :: Switch -> DialOpts -> Direction -> [(Multiaddr, Transport)]
  -> IO (Either DialError Connection)
staggeredDial :: Switch
-> DialOpts
-> Direction
-> [(Multiaddr, Transport)]
-> IO (Either DialError Connection)
staggeredDial Switch
sw DialOpts
opts Direction
dir [(Multiaddr, Transport)]
pairs =
  IO [Async Connection]
-> ([Async Connection] -> IO ())
-> ([Async Connection] -> IO (Either DialError Connection))
-> IO (Either DialError Connection)
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracketOnError IO [Async Connection]
spawnWorkers [Async Connection] -> IO ()
cancelAndCloseWorkers ([Async Connection] -> [String] -> IO (Either DialError Connection)
`collectResults` [])
  where
    -- Spawn workers with staggered delays: 0ms, 250ms, 500ms, ...
    spawnWorkers :: IO [Async Connection]
spawnWorkers = [(Int, (Multiaddr, Transport))]
-> ((Int, (Multiaddr, Transport)) -> IO (Async Connection))
-> IO [Async Connection]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM ([Int]
-> [(Multiaddr, Transport)] -> [(Int, (Multiaddr, Transport))]
forall a b. [a] -> [b] -> [(a, b)]
zip [Int
0 :: Int ..] [(Multiaddr, Transport)]
pairs) (((Int, (Multiaddr, Transport)) -> IO (Async Connection))
 -> IO [Async Connection])
-> ((Int, (Multiaddr, Transport)) -> IO (Async Connection))
-> IO [Async Connection]
forall a b. (a -> b) -> a -> b
$ \(Int
i, (Multiaddr
addr, Transport
transport)) ->
      IO Connection -> IO (Async Connection)
forall a. IO a -> IO (Async a)
async (IO Connection -> IO (Async Connection))
-> IO Connection -> IO (Async Connection)
forall a b. (a -> b) -> a -> b
$ do
        Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
i Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Int -> IO ()
threadDelay (Int
i Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
staggerDelayUs)
        localBind <- Switch -> Bool -> Multiaddr -> IO (Maybe Multiaddr)
localBindFor Switch
sw (DialOpts -> Bool
doForceDirect DialOpts
opts) Multiaddr
addr
        rawConn <- transportDialFrom transport localBind addr
        upgradeAs dir (swIdentityKey sw) rawConn
          `onException` rcClose rawConn

-- | Hole-punch dials bind the outgoing socket to a same-family listen
-- address so the SYN shares the listen port. Ordinary dials leave the
-- source port ephemeral.
localBindFor :: Switch -> Bool -> Multiaddr -> IO (Maybe Multiaddr)
localBindFor :: Switch -> Bool -> Multiaddr -> IO (Maybe Multiaddr)
localBindFor Switch
_ Bool
False Multiaddr
_ = Maybe Multiaddr -> IO (Maybe Multiaddr)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe Multiaddr
forall a. Maybe a
Nothing
localBindFor Switch
sw Bool
True Multiaddr
remote = do
  addrs <- Switch -> IO [Multiaddr]
switchListenAddrs Switch
sw
  pure $ find (sameIpFamily remote) (filter (not . isRelayedAddr) addrs)

sameIpFamily :: Multiaddr -> Multiaddr -> Bool
sameIpFamily :: Multiaddr -> Multiaddr -> Bool
sameIpFamily Multiaddr
a Multiaddr
b = Multiaddr -> Maybe Int
ipKind Multiaddr
a Maybe Int -> Maybe Int -> Bool
forall a. Eq a => a -> a -> Bool
== Multiaddr -> Maybe Int
ipKind Multiaddr
b Bool -> Bool -> Bool
&& Multiaddr -> Maybe Int
ipKind Multiaddr
a Maybe Int -> Maybe Int -> Bool
forall a. Eq a => a -> a -> Bool
/= Maybe Int
forall a. Maybe a
Nothing
  where
    ipKind :: Multiaddr -> Maybe Int
ipKind (Multiaddr (IP4 Word32
_ : [Protocol]
_)) = Int -> Maybe Int
forall a. a -> Maybe a
Just (Int
0 :: Int)
    ipKind (Multiaddr (IP6 ByteString
_ : [Protocol]
_)) = Int -> Maybe Int
forall a. a -> Maybe a
Just Int
1
    ipKind Multiaddr
_ = Maybe Int
forall a. Maybe a
Nothing

-- | Wait for the first successful async result, cancelling the rest.
-- If all fail, return DialAllFailed with all error messages.
collectResults :: [Async Connection] -> [String] -> IO (Either DialError Connection)
collectResults :: [Async Connection] -> [String] -> IO (Either DialError Connection)
collectResults [] [String]
errs = Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (DialError -> Either DialError Connection
forall a b. a -> Either a b
Left ([String] -> DialError
DialAllFailed ([String] -> [String]
forall a. [a] -> [a]
reverse [String]
errs)))
collectResults [Async Connection]
workers [String]
errs = do
  (completed, result) <- [Async Connection]
-> IO (Async Connection, Either SomeException Connection)
forall a. [Async a] -> IO (Async a, Either SomeException a)
waitAnyCatch [Async Connection]
workers
  let remaining = (Async Connection -> Bool)
-> [Async Connection] -> [Async Connection]
forall a. (a -> Bool) -> [a] -> [a]
filter (Async Connection -> Async Connection -> Bool
forall a. Eq a => a -> a -> Bool
/= Async Connection
completed) [Async Connection]
workers
  case result of
    Right Connection
conn -> do
      [Async Connection] -> IO ()
cancelAndCloseWorkers [Async Connection]
remaining
      Either DialError Connection -> IO (Either DialError Connection)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Connection -> Either DialError Connection
forall a b. b -> Either a b
Right Connection
conn)
    Left (SomeException
ex :: SomeException) ->
      [Async Connection] -> [String] -> IO (Either DialError Connection)
collectResults [Async Connection]
remaining (SomeException -> String
forall a. Show a => a -> String
show SomeException
ex String -> [String] -> [String]
forall a. a -> [a] -> [a]
: [String]
errs)

-- | Stop losing dial workers and close any connection that crossed the
-- upgrade finish line before cancellation reached it.
cancelAndCloseWorkers :: [Async Connection] -> IO ()
cancelAndCloseWorkers :: [Async Connection] -> IO ()
cancelAndCloseWorkers [Async Connection]
workers = do
  (Async Connection -> IO ()) -> [Async Connection] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ Async Connection -> IO ()
forall a. Async a -> IO ()
cancel [Async Connection]
workers
  [Async Connection] -> (Async Connection -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Async Connection]
workers ((Async Connection -> IO ()) -> IO ())
-> (Async Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Async Connection
worker -> do
    outcome <- Async Connection -> IO (Either SomeException Connection)
forall a. Async a -> IO (Either SomeException a)
waitCatch Async Connection
worker
    case outcome of
      Right Connection
conn -> MuxerSession -> IO ()
muxClose (Connection -> MuxerSession
connSession Connection
conn)
        IO () -> (SomeException -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch` \(SomeException
_ :: SomeException) -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      Left SomeException
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()