diff --git a/README.md b/README.md index b620200b..c3145770 100644 --- a/README.md +++ b/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])` diff --git a/index.js b/index.js index a964e9c3..e7706c55 100644 --- a/index.js +++ b/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() diff --git a/lib/connect.js b/lib/connect.js index 86afb4eb..83da37a2 100644 --- a/lib/connect.js +++ b/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() diff --git a/lib/relay-connection.js b/lib/relay-connection.js new file mode 100644 index 00000000..a72ceee0 --- /dev/null +++ b/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() +} diff --git a/lib/relay-pool.js b/lib/relay-pool.js new file mode 100644 index 00000000..a7757a0e --- /dev/null +++ b/lib/relay-pool.js @@ -0,0 +1,183 @@ +const { EventEmitter, once } = require('events') +const b4a = require('b4a') +const relay = require('blind-relay') + +module.exports = class RelayPool { + constructor(dht) { + this._dht = dht + this._entries = new Map() + } + + pair(publicKey, opts) { + return this._get(publicKey, opts.keepAlive).pair(opts) + } + + async destroy() { + const closing = [] + + for (const entry of this._entries.values()) { + closing.push(entry.destroy()) + } + + await Promise.allSettled(closing) + } + + _get(publicKey, keepAlive) { + const keyString = b4a.toString(publicKey, 'hex') + let entry = this._entries.get(keyString) + + if (!entry || entry.destroyed) { + entry = new RelayPoolEntry(this, keyString, publicKey, keepAlive) + this._entries.set(keyString, entry) + } else { + entry.setKeepAlive(keepAlive) + } + + return entry + } + + _delete(entry) { + if (this._entries.get(entry.keyString) === entry) this._entries.delete(entry.keyString) + } +} + +class RelayPoolEntry { + constructor(pool, keyString, publicKey, keepAlive) { + this.pool = pool + this.keyString = keyString + this.socket = pool._dht.connect(publicKey) + this.client = relay.Client.from(this.socket, { id: this.socket.publicKey }) + this.pairings = new Map() + this.destroyed = false + + this._onclose = this._onclose.bind(this) + this.socket.once('close', this._onclose) + this.socket.on('error', noop) + this.setKeepAlive(keepAlive) + } + + setKeepAlive(keepAlive) { + this.socket.setKeepAlive(keepAlive) + } + + pair({ isInitiator, token, stream }) { + const request = this.client.pair(isInitiator, token, stream) + const pairing = new RelayPoolPairing(this, token, request) + + this.pairings.set(pairing.keyString, pairing) + + return pairing + } + + release(pairing, unpair) { + if (!this.pairings.delete(pairing.keyString)) return + + if (unpair && !this.destroyed) { + try { + this.client.unpair(pairing.token) + } catch {} + } + + this._closeMaybe(unpair) + } + + close(graceful = true) { + if (this.destroyed) return + + this.destroyed = true + this.pool._delete(this) + this.socket.off('close', this._onclose) + if (graceful) this.socket.end() + else this.socket.destroy() + } + + async destroy() { + if (this.destroyed) return + + this.destroyed = true + this.pool._delete(this) + + for (const pairing of this.pairings.values()) { + pairing.detach() + } + + this.pairings.clear() + this.socket.off('close', this._onclose) + + const closed = this.socket.destroyed ? null : once(this.socket, 'close') + this.socket.destroy() + if (closed) await closed + } + + _closeMaybe(graceful) { + if (this.pairings.size === 0) this.close(graceful) + } + + _onclose() { + this.destroyed = true + this.pool._delete(this) + + for (const pairing of this.pairings.values()) { + pairing.destroy() + } + + this.pairings.clear() + } +} + +class RelayPoolPairing extends EventEmitter { + constructor(entry, token, request) { + super() + + this.entry = entry + this.token = token + this.keyString = token.toString('hex') + this.request = request + this.released = false + + this._ondata = this._ondata.bind(this) + this._onerror = this._onerror.bind(this) + + request.on('data', this._ondata) + request.on('error', this._onerror) + } + + get socket() { + return this.entry.socket + } + + release() { + this._release(true) + } + + detach() { + this._release(false) + } + + destroy() { + this._release(false) + this.request.stream.destroy() + } + + _release(unpair) { + if (this.released) return + this.released = true + + this.request.off('data', this._ondata) + this.request.off('error', this._onerror) + this.request.on('error', noop) + + this.entry.release(this, unpair) + } + + _ondata(data) { + this.emit('data', data) + } + + _onerror(err) { + if (this.listenerCount('error') === 0) return + this.emit('error', err) + } +} + +function noop() {} diff --git a/lib/relay-server.js b/lib/relay-server.js new file mode 100644 index 00000000..135d8fed --- /dev/null +++ b/lib/relay-server.js @@ -0,0 +1,116 @@ +const { EventEmitter, once } = require('events') +const { Server: BlindRelayServer } = require('blind-relay') + +module.exports = class RelayServer extends EventEmitter { + constructor(dht, opts = {}) { + super() + + this.dht = dht + this.closed = false + + this._server = dht.createServer(this._onconnection.bind(this)) + this._relay = new BlindRelayServer({ + createStream: + opts.createStream || ((streamOpts) => dht.createRawStream({ ...streamOpts, framed: true })) + }) + this._closing = null + this._serverClosed = false + + this._server.once('close', () => { + this._serverClosed = true + + if (this._closing) return + + this._closing = this._closeRelay(false) + this._closing.catch((err) => { + if (this.listenerCount('error') > 0) this.emit('error', err) + }) + }) + } + + get listening() { + return this._server.listening + } + + get publicKey() { + return this._server.publicKey + } + + get sessions() { + return this._relay.sessions + } + + get stats() { + return this._relay.stats + } + + address() { + return this._server.address() + } + + async listen(keyPair = this.dht.defaultKeyPair, opts = {}) { + await this._server.listen(keyPair, opts) + this.emit('listening') + return this + } + + close(opts = {}) { + if (this._closing) return this._closing + this._closing = this._close(!!opts.force) + return this._closing + } + + async _close(force) { + if (!this._serverClosed) { + await this._server.close() + } + + await this._closeRelay(force) + } + + async _closeRelay(force) { + if (force) await this._forceCloseRelay() + else await this._relay.close() + + if (!this.closed) { + this.closed = true + this.emit('close') + } + } + + _onconnection(socket) { + if (this.closed || this._closing) { + socket.destroy() + return + } + + const session = this._relay.accept(socket, { id: socket.remotePublicKey }) + + session.on('error', (err) => { + if (this.listenerCount('error') > 0) this.emit('error', err) + }) + + this.emit('session', session, socket) + } + + async _forceCloseRelay() { + const sessions = [...this._relay.sessions] + const closing = [] + + for (const session of sessions) { + if (!session.closed) closing.push(once(session, 'close')) + + session.destroy() + + if (session.stream && !session.stream.destroyed) { + session.stream.on('error', noop) + session.stream.destroy() + } + } + + await Promise.allSettled(closing) + await this._relay.close() + } +} + +function noop() {} diff --git a/lib/server.js b/lib/server.js index 0433c8a2..c806caad 100644 --- a/lib/server.js +++ b/lib/server.js @@ -9,6 +9,11 @@ const { FIREWALL, ERROR } = require('./constants') const { unslabbedHash } = require('./crypto') const SecurePayload = require('./secure-payload') const Holepuncher = require('./holepuncher') +const { + clearRelayTimeout, + closeRelayConnection, + destroyRelayConnection +} = require('./relay-connection') const { isPrivate } = require('bogon') const { ALREADY_LISTENING, NODE_DESTROYED, KEYPAIR_ALREADY_USED } = require('./errors') @@ -231,6 +236,7 @@ module.exports = class Server extends EventEmitter { relayToken: null, relaySocket: null, relayClient: null, + relayPairing: null, relayPaired: false } @@ -324,6 +330,9 @@ module.exports = class Server extends EventEmitter { hs.rawStream.removeListener('close', onrawstreamclose) if (hs.rawStream.connected) { + // Once this fires, changeRemote has moved the raw stream onto the direct path. + hs.rawStream.once('remote-changed', () => closeRelayConnection(hs)) + const remoteChanging = hs.rawStream.changeRemote(socket, remotePayload.udx.id, port, host) if (remoteChanging) remoteChanging.catch(safetyCatch) @@ -644,56 +653,50 @@ module.exports = class Server extends EventEmitter { } hs.relayToken = token - hs.relaySocket = this.dht.connect(publicKey) - hs.relaySocket.setKeepAlive(this.relayKeepAlive) - hs.relayClient = relay.Client.from(hs.relaySocket, { id: hs.relaySocket.publicKey }) + hs.relayPairing = this.dht._relayPool.pair(publicKey, { + isInitiator, + token, + stream: hs.rawStream, + keepAlive: this.relayKeepAlive + }) + hs.relaySocket = hs.relayPairing.socket hs.relayTimeout = setTimeout(onabort, 15000) - hs.relayClient - .pair(isInitiator, token, hs.rawStream) - .on('error', onabort) - .on('data', (remoteId) => { - if (hs.relayTimeout) clearRelayTimeout(hs) - if (hs.rawStream === null) { - onabort(null) - return - } + hs.relayPairing.on('error', onabort).on('data', (remoteId) => { + if (hs.relayTimeout) clearRelayTimeout(hs) + if (hs.rawStream === null) { + onabort(null) + return + } - hs.relayPaired = true - this.dht.stats.relaying.successes++ + hs.relayPaired = true + this.dht.stats.relaying.successes++ - if (hs.prepunching) clearTimeout(hs.prepunching) - hs.prepunching = null + if (hs.prepunching) clearTimeout(hs.prepunching) + hs.prepunching = null - const { remotePort, remoteHost, socket } = hs.relaySocket.rawStream + // Keep a stable reference since hs.relaySocket is cleared after direct upgrade. + const relaySocket = hs.relaySocket + const { remotePort, remoteHost, socket } = relaySocket.rawStream - hs.rawStream - .on('close', () => hs.relaySocket.destroy()) - .connect(socket, remoteId, remotePort, remoteHost) + hs.rawStream + .on('close', () => destroyRelayConnection(hs)) + .connect(socket, remoteId, remotePort, remoteHost) - hs.encryptedSocket = this.createSecretStream(false, hs.rawStream, { handshake: h }) + hs.encryptedSocket = this.createSecretStream(false, hs.rawStream, { handshake: h }) - this.onconnection(hs.encryptedSocket) - }) + this.onconnection(hs.encryptedSocket) + }) const dht = this.dht function onabort() { if (!hs.relayPaired) dht.stats.relaying.aborts++ - if (hs.relayTimeout) clearRelayTimeout(hs) - const socket = hs.relaySocket - hs.relayToken = null - hs.relaySocket = null - if (socket) socket.destroy() + destroyRelayConnection(hs) if (hs.aborted && hs.rawStream) hs.rawStream.destroy() } } } -function clearRelayTimeout(hs) { - clearTimeout(hs.relayTimeout) - hs.relayTimeout = null -} - function isConsistent(fw) { return fw === FIREWALL.OPEN || fw === FIREWALL.CONSISTENT } diff --git a/package.json b/package.json index 489e715b..60df5c66 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "@hyperswarm/secret-stream": "^6.6.2", "b4a": "^1.3.1", "bare-events": "^2.2.0", - "blind-relay": "^1.3.0", + "blind-relay": "^1.6.1", "bogon": "^1.0.0", "compact-encoding": "^2.4.1", "dht-rpc": "^6.15.1", diff --git a/test/relaying.js b/test/relaying.js index df0e5b86..a077db04 100644 --- a/test/relaying.js +++ b/test/relaying.js @@ -1,6 +1,8 @@ const test = require('brittle') const { once } = require('events') -const RelayServer = require('blind-relay').Server +const BlindRelay = require('blind-relay') +const RelayServer = BlindRelay.Server +const DHT = require('../') const Holepuncher = require('../lib/holepuncher') const { swarm, createDHT, endAndCloseSocket } = require('./helpers') @@ -61,6 +63,128 @@ test('relay connections through node, client side', async function (t) { await c.destroy() }) +test('relay service mode relays connections through node', async function (t) { + const { bootstrap } = await swarm(t) + + const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const c = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const lc = t.test('socket lifecycle') + lc.plan(5) + + const sessions = [] + const relayServer = b.createRelayServer() + relayServer.on('session', (session) => { + sessions.push(session) + }) + + await relayServer.listen() + + const aServer = a.createServer( + { + holepunch: false, + shareLocalAddress: false, + relayThrough: relayServer.publicKey + }, + function (socket) { + lc.pass('server socket opened') + socket + .on('data', (data) => { + lc.alike(data, Buffer.from('hello world')) + }) + .on('close', () => { + lc.pass('server socket closed') + }) + .end() + } + ) + + await aServer.listen() + + const aSocket = c.connect(aServer.publicKey, { + relayThrough: relayServer.publicKey + }) + + aSocket + .on('open', () => { + lc.pass('client socket opened') + }) + .on('close', () => { + lc.pass('client socket closed') + }) + .end('hello world') + + await lc + + t.is(sessions.length, 2, 'relay service accepts both relay transport sessions') + await waitFor(() => relayServer.stats.sessions.opened === 2) + t.is(relayServer.stats.sessions.accepted, 2, 'relay service tracks accepted sessions') + t.is(relayServer.stats.sessions.active, 2, 'relay service tracks active sessions') + + await relayServer.close() + t.ok(relayServer.closed, 'relay service closes') + t.is(relayServer.stats.sessions.active, 0, 'graceful close closes active sessions') + + await a.destroy() + await b.destroy() + await c.destroy() +}) + +test('relay service force close tears down active relayed connections', async function (t) { + const { bootstrap } = await swarm(t) + + const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const c = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const relayServer = b.createRelayServer() + await relayServer.listen() + + let serverSocket = null + const aServer = a.createServer( + { + holepunch: false, + shareLocalAddress: false, + relayThrough: relayServer.publicKey + }, + function (socket) { + serverSocket = socket + socket.on('error', noop) + socket.on('data', (data) => socket.write(data)) + } + ) + + await aServer.listen() + + const clientSocket = c.connect(aServer.publicKey, { + relayThrough: relayServer.publicKey + }) + clientSocket.on('error', noop) + + await once(clientSocket, 'open') + await waitFor(() => serverSocket !== null) + + const reply = once(clientSocket, 'data') + clientSocket.write(Buffer.from('still active')) + t.alike((await reply)[0], Buffer.from('still active'), 'relay path carries data') + await waitFor(() => relayServer.stats.streams.active === 2) + + const clientStopped = waitForSocketStopped(clientSocket) + const serverStopped = waitForSocketStopped(serverSocket) + + await relayServer.close({ force: true }) + + await Promise.all([clientStopped, serverStopped]) + + t.ok(relayServer.closed, 'relay service force closes') + t.is(relayServer.stats.streams.active, 0, 'force close tears down active relay streams') + + await a.destroy() + await b.destroy() + await c.destroy() +}) + test('relay connections through node, client side, client aborts hole punch', async function (t) { const { bootstrap } = await swarm(t) @@ -212,6 +336,37 @@ async function waitFor(fn, timeout = 2000) { } } +function getOnlyRelayPoolEntry(node) { + const entries = [...node._relayPool._entries.values()] + if (entries.length !== 1) throw new Error('Expected exactly one relay pool entry') + return entries[0] +} + +function waitForSocketStopped(socket, timeout = 2000) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup() + reject(new Error('Timed out waiting for socket to stop')) + }, timeout) + + socket.once('close', done) + socket.once('error', done) + + function done() { + cleanup() + resolve() + } + + function cleanup() { + clearTimeout(timer) + socket.off('close', done) + socket.off('error', done) + } + }) +} + +function noop() {} + test('relay connections through node, client side, server aborts hole punch', async function (t) { const { bootstrap } = await swarm(t) @@ -521,6 +676,322 @@ test('relay connections through node, client and server side', async function (t await a.destroy() }) +test('relay connections through same node reuse transport sockets', async function (t) { + const { bootstrap } = await swarm(t) + + const relayNode = createDHT({ bootstrap }) + const serverNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const clientNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const relaySockets = [] + let closedRelaySockets = 0 + let resolveRelaySockets = null + const relaySocketsOpened = new Promise((resolve) => { + resolveRelaySockets = resolve + }) + + const serverSockets = [] + let resolveServerSockets = null + const serverSocketsOpened = new Promise((resolve) => { + resolveServerSockets = resolve + }) + + const relay = new RelayServer({ + createStream(opts) { + return relayNode.createRawStream({ ...opts, framed: true }) + } + }) + + t.teardown(() => relay.close()) + + const relayStreams = [] + let closedRelayStreams = 0 + let resolveRelayStreamsPaired = null + const relayStreamsPaired = new Promise((resolve) => { + resolveRelayStreamsPaired = resolve + }) + let resolveFirstPairingReleased = null + const firstPairingReleased = new Promise((resolve) => { + resolveFirstPairingReleased = resolve + }) + + const relayTransportServer = relayNode.createServer(function (socket) { + relaySockets.push(socket) + socket.once('close', function () { + closedRelaySockets++ + }) + // Wait until both peers have opened their shared relay transport sockets. + if (relaySockets.length === 2) resolveRelaySockets() + + const session = relay.accept(socket, { id: socket.remotePublicKey }) + session.on('pair', function (_, __, stream) { + relayStreams.push(stream) + if (relayStreams.length === 4) resolveRelayStreamsPaired() + + stream.once('close', function () { + closedRelayStreams++ + if (closedRelayStreams === 2) resolveFirstPairingReleased() + }) + }) + session.on('error', (err) => t.comment(err.message)) + }) + + await relayTransportServer.listen() + + const appServer = serverNode.createServer( + { + holepunch: false, + shareLocalAddress: false, + relayThrough: relayTransportServer.publicKey + }, + function (socket) { + serverSockets.push(socket) + if (serverSockets.length === 2) resolveServerSockets() + + socket.on('data', (data) => socket.write(data)) + socket.on('end', () => socket.end()) + } + ) + + await appServer.listen() + + const clientSockets = [ + clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }), + clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }) + ] + + await Promise.all([ + relaySocketsOpened, + relayStreamsPaired, + serverSocketsOpened, + ...clientSockets.map((socket) => once(socket, 'open')) + ]) + + t.is(relaySockets.length, 2, 'both peers reuse one transport socket to the relay') + + const replies = clientSockets.map((socket) => once(socket, 'data')) + clientSockets[0].write(Buffer.from('hello 1')) + clientSockets[1].write(Buffer.from('hello 2')) + + t.alike((await replies[0])[0], Buffer.from('hello 1'), 'first relayed connection carries data') + t.alike((await replies[1])[0], Buffer.from('hello 2'), 'second relayed connection carries data') + + const firstServerClosed = Promise.race(serverSockets.map((socket) => once(socket, 'close'))) + await endAndCloseSocket(clientSockets[0]) + await firstServerClosed + await firstPairingReleased + t.is(closedRelayStreams, 2, 'closing one app connection releases only its relay pairing') + + const secondReply = once(clientSockets[1], 'data') + clientSockets[1].write(Buffer.from('still relayed')) + t.alike( + (await secondReply)[0], + Buffer.from('still relayed'), + 'second relayed connection stays usable after first closes' + ) + + await endAndCloseSocket(clientSockets[1]) + + for (const socket of serverSockets) { + if (!socket.destroyed) await once(socket, 'close') + } + + await waitFor(() => closedRelaySockets === 2) + t.is(closedRelaySockets, 2, 'shared relay transports close after all pairings close') + + const reconnectedServerSocket = waitFor(() => serverSockets.length === 3) + const reconnectedClientSocket = clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }) + + await Promise.all([once(reconnectedClientSocket, 'open'), reconnectedServerSocket]) + await waitFor(() => relaySockets.length === 4) + t.is(relaySockets.length, 4, 'reconnect creates fresh relay transport sockets') + + const reconnectedReply = once(reconnectedClientSocket, 'data') + reconnectedClientSocket.write(Buffer.from('reconnected')) + t.alike( + (await reconnectedReply)[0], + Buffer.from('reconnected'), + 'reconnected relay path carries data' + ) + + await endAndCloseSocket(reconnectedClientSocket) + const lastServerSocket = serverSockets[serverSockets.length - 1] + if (!lastServerSocket.destroyed) await once(lastServerSocket, 'close') + + await clientNode.destroy() + await serverNode.destroy() + await relayNode.destroy() +}) + +test('relay pool closes app streams when shared transports close', async function (t) { + const { bootstrap } = await swarm(t) + + const relayNode = createDHT({ bootstrap }) + const serverNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const clientNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const relaySockets = [] + let resolveRelaySockets = null + const relaySocketsOpened = new Promise((resolve) => { + resolveRelaySockets = resolve + }) + + const serverSockets = [] + let resolveServerSockets = null + const serverSocketsOpened = new Promise((resolve) => { + resolveServerSockets = resolve + }) + + const relay = new RelayServer({ + createStream(opts) { + return relayNode.createRawStream({ ...opts, framed: true }) + } + }) + + t.teardown(() => relay.close()) + + const relayTransportServer = relayNode.createServer(function (socket) { + relaySockets.push(socket) + if (relaySockets.length === 2) resolveRelaySockets() + + const session = relay.accept(socket, { id: socket.remotePublicKey }) + session.on('error', () => {}) + }) + + await relayTransportServer.listen() + + const appServer = serverNode.createServer( + { + holepunch: false, + shareLocalAddress: false, + relayThrough: relayTransportServer.publicKey + }, + function (socket) { + serverSockets.push(socket) + if (serverSockets.length === 2) resolveServerSockets() + + socket.on('data', (data) => socket.write(data)) + socket.on('end', () => socket.end()) + } + ) + + await appServer.listen() + + const clientSockets = [ + clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }), + clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }) + ] + + await Promise.all([ + relaySocketsOpened, + serverSocketsOpened, + ...clientSockets.map((socket) => once(socket, 'open')) + ]) + + t.is(clientNode._relayPool._entries.size, 1, 'client has one shared relay pool entry') + t.is(serverNode._relayPool._entries.size, 1, 'server has one shared relay pool entry') + + const closed = [ + ...clientSockets.map((socket) => once(socket, 'close')), + ...serverSockets.map((socket) => once(socket, 'close')) + ] + + getOnlyRelayPoolEntry(clientNode).socket.destroy() + getOnlyRelayPoolEntry(serverNode).socket.destroy() + + await Promise.all(closed) + t.pass('active app streams close when shared relay transports close') + t.is(clientNode._relayPool._entries.size, 0, 'client relay pool entry is removed') + t.is(serverNode._relayPool._entries.size, 0, 'server relay pool entry is removed') + + const reconnectedServerSocket = waitFor(() => serverSockets.length === 3) + const reconnectedClientSocket = clientNode.connect(appServer.publicKey, { + holepunch: false, + localConnection: false, + relayThrough: relayTransportServer.publicKey + }) + + await Promise.all([once(reconnectedClientSocket, 'open'), reconnectedServerSocket]) + await waitFor(() => relaySockets.length === 4) + t.is(relaySockets.length, 4, 'later reconnect creates fresh relay transports') + + const reply = once(reconnectedClientSocket, 'data') + reconnectedClientSocket.write(Buffer.from('fresh relay')) + t.alike((await reply)[0], Buffer.from('fresh relay'), 'fresh relay transport carries data') + + await endAndCloseSocket(reconnectedClientSocket) + const lastServerSocket = serverSockets[serverSockets.length - 1] + if (!lastServerSocket.destroyed) await once(lastServerSocket, 'close') + + await clientNode.destroy() + await serverNode.destroy() + await relayNode.destroy() +}) + +test('relay pool unpairs if pairing aborts before remote pairs', async function (t) { + const { bootstrap } = await swarm(t) + + const relayNode = createDHT({ bootstrap }) + const clientNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const relay = new RelayServer({ + createStream(opts) { + return relayNode.createRawStream({ ...opts, framed: true }) + } + }) + + t.teardown(() => relay.close()) + + const relayTransportServer = relayNode.createServer(function (socket) { + const session = relay.accept(socket, { id: socket.remotePublicKey }) + session.on('error', (err) => t.comment(err.message)) + }) + + await relayTransportServer.listen() + + const rawStream = clientNode.createRawStream({ framed: true }) + const pairing = clientNode._relayPool.pair(relayTransportServer.publicKey, { + isInitiator: true, + token: BlindRelay.token(), + stream: rawStream, + keepAlive: 5000 + }) + + await waitFor(() => relay._pairing.size === 1) + t.is(relay._pairing.size, 1, 'relay has one pending half-pairing') + + pairing.release() + rawStream.destroy() + + await waitFor(() => relay._pairing.size === 0) + await waitFor(() => clientNode._relayPool._entries.size === 0) + t.is(relay._pairing.size, 0, 'pending half-pairing is unpaired') + t.is(clientNode._relayPool._entries.size, 0, 'relay pool closes after aborted pairing') + + await clientNode.destroy() + await relayNode.destroy() +}) + test('relay connection upgrades to direct connection', async function (t) { const { bootstrap } = await swarm(t) @@ -538,7 +1009,17 @@ test('relay connection upgrades to direct connection', async function (t) { t.teardown(() => relayServer.close()) + const relaySockets = [] + let resolveRelaySockets = null + const relaySocketsOpened = new Promise((resolve) => { + resolveRelaySockets = resolve + }) + const relayTransportServer = relayNode.createServer(function (socket) { + relaySockets.push(socket) + // Wait until both client and server have opened their relay transport sockets. + if (relaySockets.length === 2) resolveRelaySockets(relaySockets) + const session = relayServer.accept(socket, { id: socket.remotePublicKey }) session.on('error', (err) => t.comment(err.message)) }) @@ -571,6 +1052,9 @@ test('relay connection upgrades to direct connection', async function (t) { }) const [serverSocket] = await Promise.all([serverSocketOpened, once(clientSocket, 'open')]) + await relaySocketsOpened + + t.is(relaySockets.length, 2, 'both peers opened relay transport sockets') t.not( serverSocket.rawStream.remotePort, @@ -590,9 +1074,12 @@ test('relay connection upgrades to direct connection', async function (t) { const clientUpgraded = once(clientSocket.rawStream, 'remote-changed') const serverUpgraded = once(serverSocket.rawStream, 'remote-changed') + const relaySocketsClosed = relaySockets.map((socket) => once(socket, 'close')) resumePunching() await Promise.all([clientUpgraded, serverUpgraded]) + await Promise.all(relaySocketsClosed) + t.pass('relay transport sockets close after direct upgrade') t.is( serverSocket.rawStream.remotePort, @@ -617,6 +1104,185 @@ test('relay connection upgrades to direct connection', async function (t) { await clientNode.destroy() }) +test('direct upgrade keeps other pooled relay pairings alive', async function (t) { + const { bootstrap } = await swarm(t) + + const relayNode = createDHT({ bootstrap }) + const serverNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + const clientNode = createDHT({ bootstrap, quickFirewall: false, ephemeral: true }) + + const resumePunching = pausePunching(t, [serverNode, clientNode]) + + const relayServer = new RelayServer({ + createStream(opts) { + return relayNode.createRawStream({ ...opts, framed: true }) + } + }) + + t.teardown(() => relayServer.close()) + + const relaySockets = [] + let closedRelaySockets = 0 + let resolveRelaySockets = null + const relaySocketsOpened = new Promise((resolve) => { + resolveRelaySockets = resolve + }) + + const relayStreams = [] + let closedRelayStreams = 0 + let resolveRelayStreamsPaired = null + const relayStreamsPaired = new Promise((resolve) => { + resolveRelayStreamsPaired = resolve + }) + + const relayTransportServer = relayNode.createServer(function (socket) { + relaySockets.push(socket) + socket.once('close', function () { + closedRelaySockets++ + }) + if (relaySockets.length === 2) resolveRelaySockets() + + const session = relayServer.accept(socket, { id: socket.remotePublicKey }) + session.on('pair', function (_, __, stream) { + relayStreams.push(stream) + if (relayStreams.length === 4) resolveRelayStreamsPaired() + + stream.once('close', function () { + closedRelayStreams++ + }) + }) + session.on('error', (err) => t.comment(err.message)) + }) + + await relayTransportServer.listen() + + let resolveUpgradeServerSocket = null + const upgradeServerSocketOpened = new Promise((resolve) => { + resolveUpgradeServerSocket = resolve + }) + const upgradeServer = serverNode.createServer( + { + relayThrough: relayTransportServer.publicKey, + shareLocalAddress: false + }, + function (socket) { + resolveUpgradeServerSocket(socket) + socket.on('data', (data) => socket.write(data)) + socket.on('end', () => socket.end()) + } + ) + + await upgradeServer.listen(DHT.keyPair()) + + let resolveRelayOnlyServerSocket = null + const relayOnlyServerSocketOpened = new Promise((resolve) => { + resolveRelayOnlyServerSocket = resolve + }) + const relayOnlyServer = serverNode.createServer( + { + relayThrough: relayTransportServer.publicKey, + shareLocalAddress: false + }, + function (socket) { + resolveRelayOnlyServerSocket(socket) + socket.on('data', (data) => socket.write(data)) + socket.on('end', () => socket.end()) + } + ) + + await relayOnlyServer.listen(DHT.keyPair()) + + const upgradeClientSocket = clientNode.connect(upgradeServer.publicKey, { + fastOpen: false, + localConnection: false + }) + const relayOnlyClientSocket = clientNode.connect(relayOnlyServer.publicKey, { + fastOpen: false, + holepunch: () => false, + localConnection: false + }) + + const [upgradeServerSocket, relayOnlyServerSocket] = await Promise.all([ + upgradeServerSocketOpened, + relayOnlyServerSocketOpened, + once(upgradeClientSocket, 'open'), + once(relayOnlyClientSocket, 'open') + ]) + await Promise.all([relaySocketsOpened, relayStreamsPaired]) + + t.is(relaySockets.length, 2, 'both app connections share the relay transport sockets') + t.is(clientNode._relayPool._entries.size, 1, 'client has one shared relay pool entry') + t.is(serverNode._relayPool._entries.size, 1, 'server has one shared relay pool entry') + + const beforeUpgrade = once(relayOnlyClientSocket, 'data') + relayOnlyClientSocket.write(Buffer.from('before direct upgrade')) + t.alike( + (await beforeUpgrade)[0], + Buffer.from('before direct upgrade'), + 'relay-only connection carries data before another pairing upgrades' + ) + + let relayOnlyClientUpgraded = false + let relayOnlyServerUpgraded = false + relayOnlyClientSocket.rawStream.once('remote-changed', () => { + relayOnlyClientUpgraded = true + }) + relayOnlyServerSocket.rawStream.once('remote-changed', () => { + relayOnlyServerUpgraded = true + }) + + const clientUpgraded = once(upgradeClientSocket.rawStream, 'remote-changed') + const serverUpgraded = once(upgradeServerSocket.rawStream, 'remote-changed') + + resumePunching() + await Promise.all([clientUpgraded, serverUpgraded]) + await waitFor(() => closedRelayStreams === 2) + + t.is(closedRelayStreams, 2, 'direct upgrade releases only its relay pairing') + t.is(closedRelaySockets, 0, 'shared relay transports stay open for the other pairing') + t.absent(relayOnlyClientUpgraded, 'relay-only client does not upgrade to direct') + t.absent(relayOnlyServerUpgraded, 'relay-only server does not upgrade to direct') + t.not( + relayOnlyServerSocket.rawStream.remotePort, + relayOnlyClientSocket.rawStream.localPort, + 'relay-only server stays on the relayed stream' + ) + t.not( + relayOnlyClientSocket.rawStream.remotePort, + relayOnlyServerSocket.rawStream.localPort, + 'relay-only client stays on the relayed stream' + ) + + const directReply = once(upgradeClientSocket, 'data') + upgradeClientSocket.write(Buffer.from('after direct upgrade')) + t.alike( + (await directReply)[0], + Buffer.from('after direct upgrade'), + 'upgraded direct connection carries data' + ) + + const relayedReply = once(relayOnlyClientSocket, 'data') + relayOnlyClientSocket.write(Buffer.from('still relayed')) + t.alike( + (await relayedReply)[0], + Buffer.from('still relayed'), + 'other pooled relay pairing stays usable after direct upgrade' + ) + + await endAndCloseSocket(upgradeClientSocket) + if (!upgradeServerSocket.destroyed) await once(upgradeServerSocket, 'close') + + await endAndCloseSocket(relayOnlyClientSocket) + if (!relayOnlyServerSocket.destroyed) await once(relayOnlyServerSocket, 'close') + + await waitFor(() => closedRelaySockets === 2) + t.is(closedRelaySockets, 2, 'shared relay transports close after remaining pairing closes') + + await relayNode.destroy() + await serverNode.destroy() + await clientNode.destroy() +}) + test.skip('relay several connections through node with pool', async function (t) { const { bootstrap } = await swarm(t)