-- | Circuit Relay v2 client transport (specs/relay/circuit-v2).
--
-- Turns a @p2p-circuit@ multiaddr into a first-class 'Transport' so that
-- relayed peers become ordinary 'Connection's in the Switch's pool.
--
-- The spec states that once the @hop@ CONNECT (dialer side) or @stop@
-- CONNECT (target side) exchange succeeds, "the original stream becomes
-- the relayed connection", which clients then upgrade "with a security
-- protocol and a multiplexer, just like they would e.g. upgrade a TCP
-- connection". This module produces the 'RawConnection' for that stream;
-- the existing upgrade pipeline in "LibP2P.Switch.Upgrade" does the rest.
--
-- Outbound: dial the relay, negotiate @hop@, send CONNECT, hand the
-- stream to the Switch as a raw connection.
--
-- Inbound: 'transportListen' reserves on a relay and registers a queue
-- keyed by that relay's peer id. The @stop@ protocol handler calls
-- 'acceptStopStream', which enqueues the relayed stream; the Switch's
-- accept loop then drives it through the normal inbound path (gating,
-- upgrade, resource limits, pool, notifiers, teardown).
module LibP2P.NAT.Relay.Transport
  ( -- * Shared state
    CircuitState
  , newCircuitState
    -- * Reservation refresh
  , ReservationRefreshConfig (..)
  , defaultReservationRefreshConfig
    -- * Transport
  , circuitTransport
    -- * Inbound relayed streams
  , acceptStopStream
    -- * Address handling (exported for testing)
  , CircuitAddr (..)
  , parseCircuitAddr
  , circuitAddrOf
  ) where

import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (async)
import Control.Concurrent.STM
  ( TQueue
  , TVar
  , atomically
  , modifyTVar'
  , newTQueue
  , newTVar
  , newTVarIO
  , readTQueue
  , readTVar
  , readTVarIO
  , writeTQueue
  , writeTVar
  )
import Control.Exception (SomeException, catch, throwIO, try)
import Control.Monad (unless)
import qualified Data.Map.Strict as Map
import Data.Time.Clock.POSIX (POSIXTime, getPOSIXTime)
import Data.Word (Word64)
import LibP2P.Crypto.PeerId (PeerId (..), peerIdBytes)
import LibP2P.Multiaddr (Multiaddr (..), fromBytes)
import LibP2P.Multiaddr.Protocol (Protocol (..))
import LibP2P.MultistreamSelect.Negotiation
  ( NegotiationResult (..)
  , StreamIO (..)
  , negotiateInitiator
  )
import LibP2P.NAT.Relay.Client (connectViaRelay, makeReservation)
import LibP2P.NAT.Relay.Message
  ( HopMessage (..)
  , RelayStatus (..)
  , Reservation (..)
  , hopProtocolId
  )
import LibP2P.Switch.ConnPool (lookupConn)
import LibP2P.Switch.Connection (newStream)
import LibP2P.Switch.Dial (dial)
import LibP2P.Switch.Listen (switchWithdrawListener)
import LibP2P.Switch.Types (Connection (..), Switch (..))
import LibP2P.Transport
  ( ConnectionEndpoint (..)
  , Listener (..)
  , RawConnection (..)
  , Transport (..)
  )

-- | A parsed circuit multiaddr.
--
-- Wire form: @\<relayTransportAddr\>\/p2p\/\<relay\>\/p2p-circuit[\/p2p\/\<target\>]@.
-- The target component is present when dialling and absent when listening.
data CircuitAddr = CircuitAddr
  { CircuitAddr -> Multiaddr
caRelayAddr :: !Multiaddr      -- ^ Relay's transport address, without the @\/p2p@ suffix
  , CircuitAddr -> PeerId
caRelayId   :: !PeerId         -- ^ Relay's peer id
  , CircuitAddr -> Maybe PeerId
caTarget    :: !(Maybe PeerId) -- ^ Destination peer id, when dialling
  } deriving (Int -> CircuitAddr -> ShowS
[CircuitAddr] -> ShowS
CircuitAddr -> [Char]
(Int -> CircuitAddr -> ShowS)
-> (CircuitAddr -> [Char])
-> ([CircuitAddr] -> ShowS)
-> Show CircuitAddr
forall a.
(Int -> a -> ShowS) -> (a -> [Char]) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> CircuitAddr -> ShowS
showsPrec :: Int -> CircuitAddr -> ShowS
$cshow :: CircuitAddr -> [Char]
show :: CircuitAddr -> [Char]
$cshowList :: [CircuitAddr] -> ShowS
showList :: [CircuitAddr] -> ShowS
Show, CircuitAddr -> CircuitAddr -> Bool
(CircuitAddr -> CircuitAddr -> Bool)
-> (CircuitAddr -> CircuitAddr -> Bool) -> Eq CircuitAddr
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: CircuitAddr -> CircuitAddr -> Bool
== :: CircuitAddr -> CircuitAddr -> Bool
$c/= :: CircuitAddr -> CircuitAddr -> Bool
/= :: CircuitAddr -> CircuitAddr -> Bool
Eq)

-- | Per-Switch state shared between the circuit transport and the @stop@
-- protocol handler: one inbound queue per relay we hold a reservation on.
newtype CircuitState = CircuitState (TVar (Map.Map PeerId InboundQueue))

-- | An inbound queue for relayed connections arriving via one relay.
-- The closed flag lets 'listenerClose' unblock a waiting 'listenerAccept'
-- so the Switch's accept loop terminates.
data InboundQueue = InboundQueue
  { InboundQueue -> TQueue RawConnection
iqQueue  :: !(TQueue RawConnection)
  , InboundQueue -> TVar Bool
iqClosed :: !(TVar Bool)
  }

-- | Create empty circuit state.
newCircuitState :: IO CircuitState
newCircuitState :: IO CircuitState
newCircuitState = TVar (Map PeerId InboundQueue) -> CircuitState
CircuitState (TVar (Map PeerId InboundQueue) -> CircuitState)
-> IO (TVar (Map PeerId InboundQueue)) -> IO CircuitState
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Map PeerId InboundQueue -> IO (TVar (Map PeerId InboundQueue))
forall a. a -> IO (TVar a)
newTVarIO Map PeerId InboundQueue
forall k a. Map k a
Map.empty

-- | Tuning for client-side reservation refresh (specs/relay/circuit-v2:
-- "the reservation becomes invalid after this time and it's the
-- responsibility of the client to refresh").
--
-- Values follow go-libp2p's @autorelay@ relay finder
-- (@rsvpExpirationSlack@ / @rsvpRefreshInterval@): refresh once the
-- reservation is within 'rrcMargin' of its expiry, checked every
-- 'rrcPollInterval'.
data ReservationRefreshConfig = ReservationRefreshConfig
  { ReservationRefreshConfig -> POSIXTime
rrcMargin       :: !POSIXTime  -- ^ Refresh once expiry is within this margin
  , ReservationRefreshConfig -> Int
rrcPollInterval :: !Int        -- ^ Microseconds between expiry checks
  } deriving (Int -> ReservationRefreshConfig -> ShowS
[ReservationRefreshConfig] -> ShowS
ReservationRefreshConfig -> [Char]
(Int -> ReservationRefreshConfig -> ShowS)
-> (ReservationRefreshConfig -> [Char])
-> ([ReservationRefreshConfig] -> ShowS)
-> Show ReservationRefreshConfig
forall a.
(Int -> a -> ShowS) -> (a -> [Char]) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ReservationRefreshConfig -> ShowS
showsPrec :: Int -> ReservationRefreshConfig -> ShowS
$cshow :: ReservationRefreshConfig -> [Char]
show :: ReservationRefreshConfig -> [Char]
$cshowList :: [ReservationRefreshConfig] -> ShowS
showList :: [ReservationRefreshConfig] -> ShowS
Show, ReservationRefreshConfig -> ReservationRefreshConfig -> Bool
(ReservationRefreshConfig -> ReservationRefreshConfig -> Bool)
-> (ReservationRefreshConfig -> ReservationRefreshConfig -> Bool)
-> Eq ReservationRefreshConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ReservationRefreshConfig -> ReservationRefreshConfig -> Bool
== :: ReservationRefreshConfig -> ReservationRefreshConfig -> Bool
$c/= :: ReservationRefreshConfig -> ReservationRefreshConfig -> Bool
/= :: ReservationRefreshConfig -> ReservationRefreshConfig -> Bool
Eq)

-- | Default refresh tuning: a 2 minute margin, checked every minute.
defaultReservationRefreshConfig :: ReservationRefreshConfig
defaultReservationRefreshConfig :: ReservationRefreshConfig
defaultReservationRefreshConfig = ReservationRefreshConfig
  { rrcMargin :: POSIXTime
rrcMargin       = POSIXTime
120
  , rrcPollInterval :: Int
rrcPollInterval = Int
60 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1000000
  }

-- | The Circuit Relay v2 client transport.
--
-- Captures the Switch so it can dial the relay; register it after
-- 'LibP2P.Switch.newSwitch' with 'LibP2P.Switch.addTransport'.
circuitTransport :: Switch -> CircuitState -> ReservationRefreshConfig -> Transport
circuitTransport :: Switch -> CircuitState -> ReservationRefreshConfig -> Transport
circuitTransport Switch
sw CircuitState
st ReservationRefreshConfig
refreshCfg = Transport
  { transportDial :: Multiaddr -> IO RawConnection
transportDial     = Switch -> Multiaddr -> IO RawConnection
dialCircuit Switch
sw
  , transportDialFrom :: Maybe Multiaddr -> Multiaddr -> IO RawConnection
transportDialFrom = \Maybe Multiaddr
_ -> Switch -> Multiaddr -> IO RawConnection
dialCircuit Switch
sw
  , transportListen :: Multiaddr -> IO Listener
transportListen   = Switch
-> CircuitState
-> ReservationRefreshConfig
-> Multiaddr
-> IO Listener
listenCircuit Switch
sw CircuitState
st ReservationRefreshConfig
refreshCfg
  , transportCanDial :: Multiaddr -> Bool
transportCanDial  = ([Char] -> Bool)
-> (CircuitAddr -> Bool) -> Either [Char] CircuitAddr -> Bool
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Bool -> [Char] -> Bool
forall a b. a -> b -> a
const Bool
False) (Bool -> CircuitAddr -> Bool
forall a b. a -> b -> a
const Bool
True) (Either [Char] CircuitAddr -> Bool)
-> (Multiaddr -> Either [Char] CircuitAddr) -> Multiaddr -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Multiaddr -> Either [Char] CircuitAddr
parseCircuitAddr
  }

-- Address handling

-- | Parse a circuit multiaddr into its relay and target parts.
parseCircuitAddr :: Multiaddr -> Either String CircuitAddr
parseCircuitAddr :: Multiaddr -> Either [Char] CircuitAddr
parseCircuitAddr (Multiaddr [Protocol]
ps) = case (Protocol -> Bool) -> [Protocol] -> ([Protocol], [Protocol])
forall a. (a -> Bool) -> [a] -> ([a], [a])
break (Protocol -> Protocol -> Bool
forall a. Eq a => a -> a -> Bool
== Protocol
P2PCircuit) [Protocol]
ps of
  ([Protocol]
_, []) -> [Char] -> Either [Char] CircuitAddr
forall a b. a -> Either a b
Left [Char]
"circuit address: no /p2p-circuit component"
  ([Protocol]
before, Protocol
_ : [Protocol]
after) -> do
    (relayAddr, relayId) <- [Protocol] -> Either [Char] (Multiaddr, PeerId)
forall {a}.
IsString a =>
[Protocol] -> Either a (Multiaddr, PeerId)
splitRelay [Protocol]
before
    target <- parseTarget after
    pure CircuitAddr
      { caRelayAddr = relayAddr
      , caRelayId   = relayId
      , caTarget    = target
      }
  where
    splitRelay :: [Protocol] -> Either a (Multiaddr, PeerId)
splitRelay [Protocol]
comps = case [Protocol] -> [Protocol]
forall a. [a] -> [a]
reverse [Protocol]
comps of
      (P2P ByteString
pid : [Protocol]
rest)
        | Bool -> Bool
not ([Protocol] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Protocol]
rest) -> (Multiaddr, PeerId) -> Either a (Multiaddr, PeerId)
forall a b. b -> Either a b
Right ([Protocol] -> Multiaddr
Multiaddr ([Protocol] -> [Protocol]
forall a. [a] -> [a]
reverse [Protocol]
rest), ByteString -> PeerId
PeerId ByteString
pid)
        | Bool
otherwise -> a -> Either a (Multiaddr, PeerId)
forall a b. a -> Either a b
Left a
"circuit address: relay has no transport address"
      [Protocol]
_ -> a -> Either a (Multiaddr, PeerId)
forall a b. a -> Either a b
Left a
"circuit address: relay component must end with /p2p/<relay>"
    parseTarget :: [Protocol] -> Either a (Maybe PeerId)
parseTarget [] = Maybe PeerId -> Either a (Maybe PeerId)
forall a b. b -> Either a b
Right Maybe PeerId
forall a. Maybe a
Nothing
    parseTarget [P2P ByteString
pid] = Maybe PeerId -> Either a (Maybe PeerId)
forall a b. b -> Either a b
Right (PeerId -> Maybe PeerId
forall a. a -> Maybe a
Just (ByteString -> PeerId
PeerId ByteString
pid))
    parseTarget [Protocol]
_ =
      a -> Either a (Maybe PeerId)
forall a b. a -> Either a b
Left a
"circuit address: expected at most /p2p/<target> after /p2p-circuit"

-- | Build the circuit multiaddr describing a relayed connection.
circuitAddrOf :: Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr
circuitAddrOf :: Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr
circuitAddrOf Multiaddr
relayAddr PeerId
relayId Maybe PeerId
mTarget =
  [Protocol] -> Multiaddr
Multiaddr (Multiaddr -> [Protocol]
stripP2P Multiaddr
relayAddr [Protocol] -> [Protocol] -> [Protocol]
forall a. [a] -> [a] -> [a]
++ [ByteString -> Protocol
P2P (PeerId -> ByteString
peerIdBytes PeerId
relayId), Protocol
P2PCircuit] [Protocol] -> [Protocol] -> [Protocol]
forall a. [a] -> [a] -> [a]
++ [Protocol]
targetPart)
  where
    targetPart :: [Protocol]
targetPart = [Protocol] -> (PeerId -> [Protocol]) -> Maybe PeerId -> [Protocol]
forall b a. b -> (a -> b) -> Maybe a -> b
maybe [] (\PeerId
t -> [ByteString -> Protocol
P2P (PeerId -> ByteString
peerIdBytes PeerId
t)]) Maybe PeerId
mTarget
    stripP2P :: Multiaddr -> [Protocol]
stripP2P (Multiaddr [Protocol]
comps) = case [Protocol] -> [Protocol]
forall a. [a] -> [a]
reverse [Protocol]
comps of
      (P2P ByteString
_ : [Protocol]
rest) -> [Protocol] -> [Protocol]
forall a. [a] -> [a]
reverse [Protocol]
rest
      [Protocol]
_              -> [Protocol]
comps

-- Outbound

-- | Dial a peer through a relay.
--
-- Connects to the relay, negotiates @hop@, sends CONNECT for the target,
-- and on @STATUS OK@ returns the hop stream as the raw relayed connection.
dialCircuit :: Switch -> Multiaddr -> IO RawConnection
dialCircuit :: Switch -> Multiaddr -> IO RawConnection
dialCircuit Switch
sw Multiaddr
addr = do
  circuit <- ([Char] -> IO CircuitAddr)
-> (CircuitAddr -> IO CircuitAddr)
-> Either [Char] CircuitAddr
-> IO CircuitAddr
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either [Char] -> IO CircuitAddr
forall a. [Char] -> IO a
forall (m :: * -> *) a. MonadFail m => [Char] -> m a
fail CircuitAddr -> IO CircuitAddr
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Multiaddr -> Either [Char] CircuitAddr
parseCircuitAddr Multiaddr
addr)
  target <- maybe (fail "circuit dial: address has no /p2p/<target>") pure
              (caTarget circuit)
  relayConn <- dialRelay sw circuit
  stream <- openHopStream sw relayConn
  resp <- connectViaRelay stream target >>= either (failClosing stream) pure
  unless (hopStatus resp == Just RelayOK) $
    failClosing stream ("relay refused CONNECT: " ++ show (hopStatus resp))
  pure RawConnection
    { rcEndpoint   = ByteStreamEndpoint stream
    , rcLocalAddr  = connLocalAddr relayConn
    , rcRemoteAddr = circuitAddrOf (caRelayAddr circuit) (caRelayId circuit) (Just target)
    , rcClose      = closeQuietly stream
    }

-- Inbound

-- | Reserve a slot on a relay and listen for relayed connections through it.
--
-- The @hop@ stream used for RESERVE is closed once the reservation is
-- granted: per the spec the reservation lives as long as the connection
-- to the relay, and inbound circuits arrive as fresh @stop@ streams on
-- that connection. A background loop re-issues RESERVE on fresh @hop@
-- streams ahead of the granted expiry to keep the reservation alive; see
-- 'refreshLoop'.
listenCircuit :: Switch -> CircuitState -> ReservationRefreshConfig -> Multiaddr -> IO Listener
listenCircuit :: Switch
-> CircuitState
-> ReservationRefreshConfig
-> Multiaddr
-> IO Listener
listenCircuit Switch
sw CircuitState
st ReservationRefreshConfig
refreshCfg Multiaddr
addr = do
  circuit <- ([Char] -> IO CircuitAddr)
-> (CircuitAddr -> IO CircuitAddr)
-> Either [Char] CircuitAddr
-> IO CircuitAddr
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either [Char] -> IO CircuitAddr
forall a. [Char] -> IO a
forall (m :: * -> *) a. MonadFail m => [Char] -> m a
fail CircuitAddr -> IO CircuitAddr
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Multiaddr -> Either [Char] CircuitAddr
parseCircuitAddr Multiaddr
addr)
  relayConn <- dialRelay sw circuit
  stream <- openHopStream sw relayConn
  resp <- makeReservation stream >>= either (failClosing stream) pure
  unless (hopStatus resp == Just RelayOK) $
    failClosing stream ("relay refused RESERVE: " ++ show (hopStatus resp))
  expiry <- either (failClosing stream) pure (reservationExpiry resp)
  closeQuietly stream
  queue <- registerQueue st (caRelayId circuit)
  let listenAddr = CircuitAddr -> HopMessage -> Multiaddr
reservationAddr CircuitAddr
circuit HopMessage
resp
  registerRelayLossNotifier sw relayConn listenAddr
  _ <- async (refreshLoop refreshCfg sw relayConn queue listenAddr expiry)
  pure Listener
    { listenerAccept = acceptFrom queue
    , listenerClose  = unregisterQueue st (caRelayId circuit)
    , listenerAddr   = listenAddr
    }

-- | Extract the expiration time the relay granted, per circuit-v2's
-- Reservation.expire: "a UTC UNIX time in seconds". Refresh cannot be
-- scheduled without it, so a missing value is treated as a failure
-- rather than defaulted.
reservationExpiry :: HopMessage -> Either String Word64
reservationExpiry :: HopMessage -> Either [Char] Word64
reservationExpiry HopMessage
resp = case HopMessage -> Maybe Reservation
hopReservation HopMessage
resp Maybe Reservation -> (Reservation -> Maybe Word64) -> Maybe Word64
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Reservation -> Maybe Word64
rsvExpire of
  Just Word64
expiry -> Word64 -> Either [Char] Word64
forall a b. b -> Either a b
Right Word64
expiry
  Maybe Word64
Nothing     -> [Char] -> Either [Char] Word64
forall a b. a -> Either a b
Left [Char]
"relay RESERVE response is missing the reservation expiry"

-- | Periodically re-issue RESERVE on the existing connection to the
-- relay, ahead of the current reservation's expiry (specs/relay/circuit-v2:
-- "it's the responsibility of the client to refresh").
--
-- Stops once the listener's queue is closed -- by 'listenerClose'
-- (explicit close or 'LibP2P.Switch.switchClose'), by the relay-loss
-- notifier registered in 'listenCircuit', or by this loop itself
-- withdrawing the listener after a failed refresh.
refreshLoop
  :: ReservationRefreshConfig -> Switch -> Connection -> InboundQueue
  -> Multiaddr -> Word64 -> IO ()
refreshLoop :: ReservationRefreshConfig
-> Switch
-> Connection
-> InboundQueue
-> Multiaddr
-> Word64
-> IO ()
refreshLoop ReservationRefreshConfig
cfg Switch
sw Connection
relayConn InboundQueue
queue Multiaddr
listenAddr = Word64 -> IO ()
go
  where
    go :: Word64 -> IO ()
go Word64
expiry = do
      Int -> IO ()
threadDelay (ReservationRefreshConfig -> Int
rrcPollInterval ReservationRefreshConfig
cfg)
      closed <- TVar Bool -> IO Bool
forall a. TVar a -> IO a
readTVarIO (InboundQueue -> TVar Bool
iqClosed InboundQueue
queue)
      unless closed $ do
        now <- getPOSIXTime
        if now + rrcMargin cfg >= fromIntegral expiry
          then do
            result <- refreshReservation sw relayConn
            case result of
              Left [Char]
_err       -> Switch -> Multiaddr -> IO ()
switchWithdrawListener Switch
sw Multiaddr
listenAddr
              Right Word64
newExpiry -> Word64 -> IO ()
go Word64
newExpiry
          else go expiry

-- | Re-issue RESERVE on a fresh @hop@ stream over an existing connection
-- to the relay, returning the new expiry or a description of why the
-- refresh failed (transport error or a non-OK STATUS).
refreshReservation :: Switch -> Connection -> IO (Either String Word64)
refreshReservation :: Switch -> Connection -> IO (Either [Char] Word64)
refreshReservation Switch
sw Connection
relayConn = do
  result <- IO (Either [Char] Word64)
-> IO (Either SomeException (Either [Char] Word64))
forall e a. Exception e => IO a -> IO (Either e a)
try IO (Either [Char] Word64)
attempt
  pure $ case result of
    Left (SomeException
e :: SomeException) -> [Char] -> Either [Char] Word64
forall a b. a -> Either a b
Left (SomeException -> [Char]
forall a. Show a => a -> [Char]
show SomeException
e)
    Right Either [Char] Word64
expiry               -> Either [Char] Word64
expiry
  where
    attempt :: IO (Either [Char] Word64)
attempt = do
      stream <- Switch -> Connection -> IO StreamIO
openHopStream Switch
sw Connection
relayConn
      resp <- makeReservation stream >>= either (failClosing stream) pure
      unless (hopStatus resp == Just RelayOK) $
        failClosing stream ("relay refused RESERVE refresh: " ++ show (hopStatus resp))
      closeQuietly stream
      pure (reservationExpiry resp)

-- | Withdraw the circuit listen address once the last connection to the
-- relay is gone (specs/relay/circuit-v2: "the reservation remains valid
-- until its expiration, as long as there is an active connection from
-- the peer to the relay. If the peer disconnects, the reservation is no
-- longer valid").
--
-- The reservation is bound to the relay peer, not to the connection the
-- RESERVE went out on, so a second connection to the same relay keeps
-- the listen address alive. This is the same predicate the relay server
-- side applies in 'LibP2P.NAT.registerReservationCleanup', and it has to
-- match: against a go-libp2p relay, which keeps a reservation while any
-- connection from us remains, per-connection matching would withdraw a
-- listen address the relay still honours.
--
-- 'closeConnection' removes the connection from the pool in the same STM
-- transaction that marks it closed and only then runs the notifiers, so
-- the lookup below never observes the connection being torn down.
registerRelayLossNotifier :: Switch -> Connection -> Multiaddr -> IO ()
registerRelayLossNotifier :: Switch -> Connection -> Multiaddr -> IO ()
registerRelayLossNotifier Switch
sw Connection
relayConn Multiaddr
listenAddr =
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar [Connection -> IO ()]
-> ([Connection -> IO ()] -> [Connection -> IO ()]) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (Switch -> TVar [Connection -> IO ()]
swDisconnectNotifiers Switch
sw) (Connection -> IO ()
notifier (Connection -> IO ())
-> [Connection -> IO ()] -> [Connection -> IO ()]
forall a. a -> [a] -> [a]
:)
  where
    relayId :: PeerId
relayId = Connection -> PeerId
connPeerId Connection
relayConn
    notifier :: Connection -> IO ()
notifier Connection
conn
      | Connection -> PeerId
connPeerId Connection
conn PeerId -> PeerId -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerId
relayId = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      | Bool
otherwise = do
          remaining <- STM (Maybe Connection) -> IO (Maybe Connection)
forall a. STM a -> IO a
atomically (STM (Maybe Connection) -> IO (Maybe Connection))
-> STM (Maybe Connection) -> IO (Maybe Connection)
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId [Connection]) -> PeerId -> STM (Maybe Connection)
lookupConn (Switch -> TVar (Map PeerId [Connection])
swConnPool Switch
sw) PeerId
relayId
          case remaining of
            Just Connection
_  -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            Maybe Connection
Nothing -> Switch -> Multiaddr -> IO ()
switchWithdrawListener Switch
sw Multiaddr
listenAddr

-- | Hand a relayed stream that arrived via the @stop@ protocol to the
-- listener for the relay it came over.
--
-- The stream is closed when no listener is registered for that relay:
-- without a reservation we have nothing to accept the circuit into.
acceptStopStream :: CircuitState -> Connection -> PeerId -> StreamIO -> IO ()
acceptStopStream :: CircuitState -> Connection -> PeerId -> StreamIO -> IO ()
acceptStopStream (CircuitState TVar (Map PeerId InboundQueue)
var) Connection
relayConn PeerId
source StreamIO
stream = do
  enqueued <- 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
$ do
    queues <- TVar (Map PeerId InboundQueue) -> STM (Map PeerId InboundQueue)
forall a. TVar a -> STM a
readTVar TVar (Map PeerId InboundQueue)
var
    case Map.lookup (connPeerId relayConn) queues of
      Maybe InboundQueue
Nothing -> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
      Just InboundQueue
q -> do
        closed <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar (InboundQueue -> TVar Bool
iqClosed InboundQueue
q)
        if closed
          then pure False
          else do
            writeTQueue (iqQueue q) rawConn
            pure True
  unless enqueued (closeQuietly stream)
  where
    rawConn :: RawConnection
rawConn = RawConnection
      { rcEndpoint :: ConnectionEndpoint
rcEndpoint   = StreamIO -> ConnectionEndpoint
ByteStreamEndpoint StreamIO
stream
      , rcLocalAddr :: Multiaddr
rcLocalAddr  = Connection -> Multiaddr
connLocalAddr Connection
relayConn
      , rcRemoteAddr :: Multiaddr
rcRemoteAddr =
          Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr
circuitAddrOf (Connection -> Multiaddr
connRemoteAddr Connection
relayConn) (Connection -> PeerId
connPeerId Connection
relayConn) (PeerId -> Maybe PeerId
forall a. a -> Maybe a
Just PeerId
source)
      , rcClose :: IO ()
rcClose      = StreamIO -> IO ()
closeQuietly StreamIO
stream
      }

-- Helpers

-- | Dial the relay named by a circuit address, reusing a pooled
-- connection to it when one exists.
dialRelay :: Switch -> CircuitAddr -> IO Connection
dialRelay :: Switch -> CircuitAddr -> IO Connection
dialRelay Switch
sw CircuitAddr
circuit =
  Switch -> PeerId -> [Multiaddr] -> IO (Either DialError Connection)
dial Switch
sw (CircuitAddr -> PeerId
caRelayId CircuitAddr
circuit) [CircuitAddr -> Multiaddr
caRelayAddr CircuitAddr
circuit]
    IO (Either DialError Connection)
-> (Either DialError Connection -> IO Connection) -> IO Connection
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (DialError -> IO Connection)
-> (Connection -> IO Connection)
-> Either DialError Connection
-> IO Connection
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (\DialError
err -> [Char] -> IO Connection
forall a. [Char] -> IO a
forall (m :: * -> *) a. MonadFail m => [Char] -> m a
fail ([Char]
"circuit: cannot reach relay: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ DialError -> [Char]
forall a. Show a => a -> [Char]
show DialError
err)) Connection -> IO Connection
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure

-- | Open a stream to the relay and negotiate the @hop@ protocol.
openHopStream :: Switch -> Connection -> IO StreamIO
openHopStream :: Switch -> Connection -> IO StreamIO
openHopStream Switch
sw Connection
relayConn = do
  stream <- Switch -> Connection -> IO (Either ResourceError StreamIO)
newStream Switch
sw Connection
relayConn
    IO (Either ResourceError StreamIO)
-> (Either ResourceError StreamIO -> IO StreamIO) -> IO StreamIO
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (ResourceError -> IO StreamIO)
-> (StreamIO -> IO StreamIO)
-> Either ResourceError StreamIO
-> IO StreamIO
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (\ResourceError
err -> [Char] -> IO StreamIO
forall a. [Char] -> IO a
forall (m :: * -> *) a. MonadFail m => [Char] -> m a
fail ([Char]
"circuit: cannot open hop stream: " [Char] -> ShowS
forall a. [a] -> [a] -> [a]
++ ResourceError -> [Char]
forall a. Show a => a -> [Char]
show ResourceError
err)) StreamIO -> IO StreamIO
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
  negotiated <- negotiateInitiator stream [hopProtocolId]
  case negotiated of
    Accepted ProtocolId
_ -> StreamIO -> IO StreamIO
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure StreamIO
stream
    NegotiationResult
NoProtocol -> StreamIO -> [Char] -> IO StreamIO
forall a. StreamIO -> [Char] -> IO a
failClosing StreamIO
stream [Char]
"relay does not support /libp2p/circuit/relay/0.2.0/hop"

-- | The address this listener is reachable on: the relay's advertised
-- reservation address with @\/p2p-circuit@ appended. Falls back to the
-- dialled relay address when the relay advertises none.
reservationAddr :: CircuitAddr -> HopMessage -> Multiaddr
reservationAddr :: CircuitAddr -> HopMessage -> Multiaddr
reservationAddr CircuitAddr
circuit HopMessage
resp =
  case HopMessage -> Maybe Reservation
hopReservation HopMessage
resp Maybe Reservation
-> (Reservation -> Maybe Multiaddr) -> Maybe Multiaddr
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [ByteString] -> Maybe Multiaddr
firstDecodable ([ByteString] -> Maybe Multiaddr)
-> (Reservation -> [ByteString]) -> Reservation -> Maybe Multiaddr
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Reservation -> [ByteString]
rsvAddrs of
    Just Multiaddr
relayAddr -> Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr
circuitAddrOf Multiaddr
relayAddr (CircuitAddr -> PeerId
caRelayId CircuitAddr
circuit) Maybe PeerId
forall a. Maybe a
Nothing
    Maybe Multiaddr
Nothing -> Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr
circuitAddrOf (CircuitAddr -> Multiaddr
caRelayAddr CircuitAddr
circuit) (CircuitAddr -> PeerId
caRelayId CircuitAddr
circuit) Maybe PeerId
forall a. Maybe a
Nothing
  where
    firstDecodable :: [ByteString] -> Maybe Multiaddr
firstDecodable [] = Maybe Multiaddr
forall a. Maybe a
Nothing
    firstDecodable (ByteString
bs : [ByteString]
rest) = ([Char] -> Maybe Multiaddr)
-> (Multiaddr -> Maybe Multiaddr)
-> Either [Char] Multiaddr
-> Maybe Multiaddr
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Maybe Multiaddr -> [Char] -> Maybe Multiaddr
forall a b. a -> b -> a
const ([ByteString] -> Maybe Multiaddr
firstDecodable [ByteString]
rest)) Multiaddr -> Maybe Multiaddr
forall a. a -> Maybe a
Just (ByteString -> Either [Char] Multiaddr
fromBytes ByteString
bs)

-- | Register an inbound queue for a relay, replacing any previous one.
registerQueue :: CircuitState -> PeerId -> IO InboundQueue
registerQueue :: CircuitState -> PeerId -> IO InboundQueue
registerQueue (CircuitState TVar (Map PeerId InboundQueue)
var) PeerId
relayId = STM InboundQueue -> IO InboundQueue
forall a. STM a -> IO a
atomically (STM InboundQueue -> IO InboundQueue)
-> STM InboundQueue -> IO InboundQueue
forall a b. (a -> b) -> a -> b
$ do
  queue <- TQueue RawConnection -> TVar Bool -> InboundQueue
InboundQueue (TQueue RawConnection -> TVar Bool -> InboundQueue)
-> STM (TQueue RawConnection) -> STM (TVar Bool -> InboundQueue)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM (TQueue RawConnection)
forall a. STM (TQueue a)
newTQueue STM (TVar Bool -> InboundQueue)
-> STM (TVar Bool) -> STM InboundQueue
forall a b. STM (a -> b) -> STM a -> STM b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Bool -> STM (TVar Bool)
forall a. a -> STM (TVar a)
newTVar Bool
False
  queues <- readTVar var
  writeTVar var (Map.insert relayId queue queues)
  pure queue

-- | Mark a relay's inbound queue closed and drop it, releasing any
-- 'listenerAccept' blocked on it.
unregisterQueue :: CircuitState -> PeerId -> IO ()
unregisterQueue :: CircuitState -> PeerId -> IO ()
unregisterQueue (CircuitState TVar (Map PeerId InboundQueue)
var) PeerId
relayId = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  queues <- TVar (Map PeerId InboundQueue) -> STM (Map PeerId InboundQueue)
forall a. TVar a -> STM a
readTVar TVar (Map PeerId InboundQueue)
var
  case Map.lookup relayId queues of
    Maybe InboundQueue
Nothing -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    Just InboundQueue
q -> do
      TVar Bool -> Bool -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (InboundQueue -> TVar Bool
iqClosed InboundQueue
q) Bool
True
      TVar (Map PeerId InboundQueue) -> Map PeerId InboundQueue -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar (Map PeerId InboundQueue)
var (PeerId -> Map PeerId InboundQueue -> Map PeerId InboundQueue
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
relayId Map PeerId InboundQueue
queues)

-- | Block for the next relayed connection, or throw once the listener is
-- closed so the Switch's accept loop stops.
acceptFrom :: InboundQueue -> IO RawConnection
acceptFrom :: InboundQueue -> IO RawConnection
acceptFrom InboundQueue
q = do
  result <- STM (Maybe RawConnection) -> IO (Maybe RawConnection)
forall a. STM a -> IO a
atomically (STM (Maybe RawConnection) -> IO (Maybe RawConnection))
-> STM (Maybe RawConnection) -> IO (Maybe RawConnection)
forall a b. (a -> b) -> a -> b
$ do
    closed <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar (InboundQueue -> TVar Bool
iqClosed InboundQueue
q)
    if closed
      then pure Nothing
      else Just <$> readTQueue (iqQueue q)
  maybe (fail "circuit listener closed") pure result

-- | Close a stream, ignoring failures from an already-dead session.
closeQuietly :: StreamIO -> IO ()
closeQuietly :: StreamIO -> IO ()
closeQuietly StreamIO
stream = StreamIO -> IO ()
streamClose StreamIO
stream 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 ()

-- | Abort with an error, closing the stream we were using first.
failClosing :: StreamIO -> String -> IO a
failClosing :: forall a. StreamIO -> [Char] -> IO a
failClosing StreamIO
stream [Char]
msg = do
  StreamIO -> IO ()
closeQuietly StreamIO
stream
  IOError -> IO a
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO ([Char] -> IOError
userError [Char]
msg)