nim-libp2p/libp2p/stream/connection.nim

# Nim-LibP2P
# Copyright (c) 2023-2024 Status Research & Development GmbH
# Licensed under either of
#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
# at your option.
# This file may not be copied, modified, or distributed except according to
# those terms.

{.push raises: [].}

import std/[hashes, oids, strformat]
import stew/results
import chronicles, chronos, metrics
import lpstream, ../multiaddress, ../peerinfo, ../errors

export lpstream, peerinfo, errors, results

logScope:
  topics = "libp2p connection"

const
  ConnectionTrackerName* = "Connection"
  DefaultConnectionTimeout* = 5.minutes

type
  TimeoutHandler* = proc(): Future[void] {.async: (raises: []).}

  Connection* = ref object of LPStream
    activity*: bool # reset every time data is sent or received
    timeout*: Duration # channel timeout if no activity
    timerTaskFut: Future[void].Raising([]) # the current timer instance
    timeoutHandler*: TimeoutHandler # timeout handler
    peerId*: PeerId
    observedAddr*: Opt[MultiAddress]
    protocol*: string # protocol used by the connection, used as metrics tag
    transportDir*: Direction # underlying transport (usually socket) direction
    when defined(libp2p_agents_metrics):
      shortAgent*: string

proc timeoutMonitor(s: Connection) {.async: (raises: []).}

func shortLog*(conn: Connection): string =
  try:
    if conn == nil:
      "Connection(nil)"
    else:
      &"{shortLog(conn.peerId)}:{conn.oid}"
  except ValueError as exc:
    raiseAssert(exc.msg)

chronicles.formatIt(Connection):
  shortLog(it)

method initStream*(s: Connection) =
  if s.objName.len == 0:
    s.objName = ConnectionTrackerName

  procCall LPStream(s).initStream()

  doAssert(s.timerTaskFut == nil)

  if s.timeout > 0.millis:
    trace "Monitoring for timeout", s, timeout = s.timeout

    s.timerTaskFut = s.timeoutMonitor()
    if s.timeoutHandler == nil:
      s.timeoutHandler = proc(): Future[void] {.async: (raises: [], raw: true).} =
        trace "Idle timeout expired, closing connection", s
        s.close()

method closeImpl*(s: Connection): Future[void] {.async: (raises: []).} =
  # Cleanup timeout timer
  trace "Closing connection", s

  if s.timerTaskFut != nil and not s.timerTaskFut.finished:
    # Don't `cancelAndWait` here to avoid risking deadlock in this scenario:
    # - `pollActivity` is waiting for `s.timeoutHandler` to complete.
    # - `s.timeoutHandler` may have triggered `closeImpl` and we are now here.
    # In this situation, we have to return for `s.timerTaskFut` to complete.
    s.timerTaskFut.cancelSoon()
    s.timerTaskFut = nil

  trace "Closed connection", s

  procCall LPStream(s).closeImpl()

func hash*(p: Connection): Hash =
  cast[pointer](p).hash

proc pollActivity(s: Connection): Future[bool] {.async: (raises: []).} =
  if s.closed and s.atEof:
    return false # Done, no more monitoring

  if s.activity:
    s.activity = false
    return true

  # Inactivity timeout happened, call timeout monitor

  trace "Connection timed out", s
  if s.timeoutHandler != nil:
    trace "Calling timeout handler", s
    await s.timeoutHandler()

  return false

proc timeoutMonitor(s: Connection) {.async: (raises: []).} =
  ## monitor the channel for inactivity
  ##
  ## if the timeout was hit, it means that
  ## neither incoming nor outgoing activity
  ## has been detected and the channel will
  ## be reset
  ##

  while true:
    try: # Sleep at least once!
      await sleepAsync(s.timeout)
    except CancelledError:
      return

    if not await s.pollActivity():
      return

method getWrapped*(s: Connection): Connection {.base.} =
  raiseAssert("Not implemented!")

when defined(libp2p_agents_metrics):
  proc setShortAgent*(s: Connection, shortAgent: string) =
    var conn = s
    while conn != nil:
      conn.shortAgent = shortAgent
      conn = conn.getWrapped()

proc new*(
    C: type Connection,
    peerId: PeerId,
    dir: Direction,
    observedAddr: Opt[MultiAddress],
    timeout: Duration = DefaultConnectionTimeout,
    timeoutHandler: TimeoutHandler = nil,
): Connection =
  result = C(
    peerId: peerId,
    dir: dir,
    timeout: timeout,
    timeoutHandler: timeoutHandler,
    observedAddr: observedAddr,
  )

  result.initStream()