Feat/relay service mode

README.md

@@ -166,6 +166,71 @@ Stop listening.
 
 Emitted when the server is fully closed.
 
+## Creating relay servers
+
+#### `const relay = node.createRelayServer([options])`
+
+Create a relay server that accepts blind relay transport connections.
+
+This is an explicit opt-in service. Peers still need to choose it with `relayThrough`.
+
+```js
+const relay = node.createRelayServer()
+await relay.listen()
+
+const socket = otherNode.connect(server.publicKey, {
+  relayThrough: relay.publicKey
+})
+```
+
+#### `await relay.listen([keyPair])`
+
+Start accepting relay transport connections using the key pair's public key.
+
+#### `relay.publicKey`
+
+The public key peers can pass as `relayThrough`.
+
+#### `relay.on('session', session, socket)`
+
+Emitted when a peer opens a relay transport session.
+
+#### `relay.sessions`
+
+An iterator over active blind relay transport sessions.
+
+#### `relay.stats`
+
+In-memory relay service counters from `blind-relay`:
+
+```js
+{
+  sessions: { accepted, opened, closed, active },
+  pairings: { requested, matched, cancelled, pending, active },
+  streams: { opened, closed, errors, active }
+}
+```
+
+#### `await relay.close([options])`
+
+Stop accepting relay transport connections and gracefully close active relay
+sessions.
+
+Graceful close uses the blind-relay protocol close path and waits for active
+relay sessions to close. It does not drain relayed application streams
+indefinitely; those streams close when their relay sessions close.
+
+Options include:
+
+```js
+{
+  force: false // use true to destroy active relay sessions immediately
+}
+```
+
+Use `await relay.close({ force: true })` when an app/user disables relay mode and
+active relayed connections should be torn down now.
+
 ## Connecting to P2P servers
 
 #### `const socket = node.connect(remotePublicKey, [options])`

index.js

@@ -9,11 +9,13 @@ const Persistent = require('./lib/persistent')
 const Router = require('./lib/router')
 const Cache = require('xache')
 const Server = require('./lib/server')
+const RelayServer = require('./lib/relay-server')
 const connect = require('./lib/connect')
 const { FIREWALL, BOOTSTRAP_NODES, KNOWN_NODES, COMMANDS } = require('./lib/constants')
 const { hash, createKeyPair } = require('./lib/crypto')
 const RawStreamSet = require('./lib/raw-stream-set')
 const ConnectionPool = require('./lib/connection-pool')
+const RelayPool = require('./lib/relay-pool')
 const { STREAM_NOT_CONNECTED } = require('./lib/errors')
 
 const DEFAULTS = {
@@ -47,6 +49,7 @@ class HyperDHT extends DHT {
     }
     this.rawStreams = new RawStreamSet(this)
     this.plugins = new Map()
+    this._relayPool = new RelayPool(this)
 
     this._router = new Router(this, router)
     this._socketPool = new SocketPool(this, opts.host || '0.0.0.0')
@@ -90,6 +93,10 @@ class HyperDHT extends DHT {
     return s
   }
 
+  createRelayServer(opts) {
+    return new RelayServer(this, opts)
+  }
+
   pool() {
     return new ConnectionPool(this)
   }
@@ -129,6 +136,7 @@ class HyperDHT extends DHT {
     this._router.destroy()
     if (this._persistent) this._persistent.destroy()
     for (const plugin of this.plugins.values()) plugin.destroy()
+    await this._relayPool.destroy()
     await this.rawStreams.clear()
     await this._socketPool.destroy()
     await super.destroy()

lib/connect.js

@@ -9,6 +9,11 @@ const NoiseWrap = require('./noise-wrap')
 const SecurePayload = require('./secure-payload')
 const Holepuncher = require('./holepuncher')
 const Sleeper = require('./sleeper')
+const {
+  clearRelayTimeout,
+  closeRelayConnection,
+  destroyRelayConnection
+} = require('./relay-connection')
 const { FIREWALL, ERROR } = require('./constants')
 const { unslabbedHash } = require('./crypto')
 const {
@@ -96,6 +101,7 @@ module.exports = function connect(dht, publicKey, opts = {}) {
     relayToken: relayThrough ? relay.token() : null,
     relaySocket: null,
     relayClient: null,
+    relayPairing: null,
     relayPaired: false,
     relayKeepAlive: opts.relayKeepAlive || 5000
   }
@@ -483,6 +489,9 @@ async function connectThroughNode(c, address, socket) {
     if (c.rawStream === null) return // Already hole punched
 
     if (c.rawStream.connected) {
+      // Once this fires, changeRemote has moved the raw stream onto the direct path.
+      c.rawStream.once('remote-changed', () => closeRelayConnection(c))
+
       const remoteChanging = c.rawStream.changeRemote(socket, c.connect.payload.udx.id, port, host)
 
       if (remoteChanging) remoteChanging.catch(safetyCatch)
@@ -790,12 +799,16 @@ function relayConnection(c, relayThrough, payload, hs) {
   }
 
   c.relayToken = token
-  c.relaySocket = c.dht.connect(publicKey)
-  c.relaySocket.setKeepAlive(c.relayKeepAlive)
-  c.relayClient = relay.Client.from(c.relaySocket, { id: c.relaySocket.publicKey })
+  c.relayPairing = c.dht._relayPool.pair(publicKey, {
+    isInitiator,
+    token,
+    stream: c.rawStream,
+    keepAlive: c.relayKeepAlive
+  })
+  c.relaySocket = c.relayPairing.socket
   c.relayTimeout = setTimeout(onabort, 15000, null)
 
-  c.relayClient.pair(isInitiator, token, c.rawStream).on('error', onabort).on('data', ondata)
+  c.relayPairing.on('error', onabort).on('data', ondata)
 
   function ondata(remoteId) {
     if (c.relayTimeout) clearRelayTimeout(c)
@@ -806,21 +819,19 @@ function relayConnection(c, relayThrough, payload, hs) {
 
     c.relayPaired = true
 
-    const { remotePort, remoteHost, socket } = c.relaySocket.rawStream
+    // Keep a stable reference since c.relaySocket is cleared after direct upgrade.
+    const relaySocket = c.relaySocket
+    const { remotePort, remoteHost, socket } = relaySocket.rawStream
 
     c.rawStream
-      .on('close', () => c.relaySocket.destroy())
+      .on('close', () => destroyRelayConnection(c))
       .connect(socket, remoteId, remotePort, remoteHost)
 
     c.encryptedSocket.start(c.rawStream, { handshake: hs })
   }
 
   function onabort(err) {
-    if (c.relayTimeout) clearRelayTimeout(c)
-    const socket = c.relaySocket
-    c.relayToken = null
-    c.relaySocket = null
-    if (socket) socket.destroy()
+    destroyRelayConnection(c)
     maybeDestroyEncryptedSocket(c, err || RELAY_ABORTED())
   }
 }
@@ -830,11 +841,6 @@ function clearPassiveConnectTimeout(c) {
   c.passiveConnectTimeout = null
 }
 
-function clearRelayTimeout(c) {
-  clearTimeout(c.relayTimeout)
-  c.relayTimeout = null
-}
-
 function destroyPuncher(c) {
   if (c.puncher) c.puncher.destroy()
   c.session.destroy()

lib/relay-connection.js

@@ -0,0 +1,43 @@
+exports.clearRelayTimeout = clearRelayTimeout
+exports.closeRelayConnection = closeRelayConnection
+exports.destroyRelayConnection = destroyRelayConnection
+
+function clearRelayTimeout(connectionOrHandshake) {
+  clearTimeout(connectionOrHandshake.relayTimeout)
+  connectionOrHandshake.relayTimeout = null
+}
+
+function closeRelayConnection(connectionOrHandshake) {
+  const relay = resetRelayConnection(connectionOrHandshake)
+
+  if (relay) closeRelay(relay)
+}
+
+function destroyRelayConnection(connectionOrHandshake) {
+  const relay = resetRelayConnection(connectionOrHandshake)
+
+  if (relay) destroyRelay(relay)
+}
+
+function resetRelayConnection(connectionOrHandshake) {
+  if (connectionOrHandshake.relayTimeout) clearRelayTimeout(connectionOrHandshake)
+
+  const relay = connectionOrHandshake.relayPairing || connectionOrHandshake.relaySocket
+  // Drop relay references so the app connection no longer keeps relay state alive.
+  connectionOrHandshake.relayToken = null
+  connectionOrHandshake.relaySocket = null
+  connectionOrHandshake.relayClient = null
+  connectionOrHandshake.relayPairing = null
+
+  return relay
+}
+
+function closeRelay(relay) {
+  if (relay.release) return relay.release()
+  relay.end()
+}
+
+function destroyRelay(relay) {
+  if (relay.release) return relay.release()
+  relay.destroy()
+}