net: add ability to reset a tcp socket

doc/api/net.md

@@ -1089,6 +1089,19 @@ added: v0.5.10
 
 The numeric representation of the remote port. For example, `80` or `21`.
 
+### `socket.resetAndDestroy()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Returns: {net.Socket}
+
+Close the TCP connection by sending an RST packet and destroy the stream.
+If this TCP socket is in connecting status, it will send an RST packet and destroy this TCP socket once it is connected.
+Otherwise, it will call `socket.destroy` with an `ERR_SOCKET_CLOSED` Error.
+If this is not a TCP socket (for example, a pipe), calling this method will immediately throw an `ERR_INVALID_HANDLE_TYPE` Error.
+
 ### `socket.resume()`
 
 * Returns: {net.Socket} The socket itself.

lib/net.js

@@ -89,6 +89,7 @@ const {
     ERR_INVALID_ARG_VALUE,
     ERR_INVALID_FD_TYPE,
     ERR_INVALID_IP_ADDRESS,
+    ERR_INVALID_HANDLE_TYPE,
     ERR_SERVER_ALREADY_LISTEN,
     ERR_SERVER_NOT_RUNNING,
     ERR_SOCKET_CLOSED,
@@ -640,6 +641,21 @@ Socket.prototype.end = function(data, encoding, callback) {
   return this;
 };
 
+Socket.prototype.resetAndDestroy = function() {
+  if (this._handle) {
+    if (!(this._handle instanceof TCP))
+      throw new ERR_INVALID_HANDLE_TYPE();
+    if (this.connecting) {
+      debug('reset wait for connection');
+      this.once('connect', () => this._reset());
+    } else {
+      this._reset();
+    }
+  } else {
+    this.destroy(new ERR_SOCKET_CLOSED());
+  }
+  return this;
+};
 
 Socket.prototype.pause = function() {
   if (this[kBuffer] && !this.connecting && this._handle &&
@@ -710,10 +726,20 @@ Socket.prototype._destroy = function(exception, cb) {
     this[kBytesRead] = this._handle.bytesRead;
     this[kBytesWritten] = this._handle.bytesWritten;
 
-    this._handle.close(() => {
-      debug('emit close');
-      this.emit('close', isException);
-    });
+    if (this.resetAndClosing) {
+      this.resetAndClosing = false;
+      const err = this._handle.reset(() => {
+        debug('emit close');
+        this.emit('close', isException);
+      });
+      if (err)
+        this.emit('error', errnoException(err, 'reset'));
+    } else {
+      this._handle.close(() => {
+        debug('emit close');
+        this.emit('close', isException);
+      });
+    }
     this._handle.onread = noop;
     this._handle = null;
     this._sockname = null;
@@ -732,6 +758,12 @@ Socket.prototype._destroy = function(exception, cb) {
   }
 };
 
+Socket.prototype._reset = function() {
+  debug('reset connection');
+  this.resetAndClosing = true;
+  return this.destroy();
+};
+
 Socket.prototype._getpeername = function() {
   if (!this._handle || !this._handle.getpeername) {
     return this._peername || {};

src/handle_wrap.h

@@ -97,6 +97,7 @@ class HandleWrap : public AsyncWrap {
   }
 
   static void OnClose(uv_handle_t* handle);
+  enum { kInitialized, kClosing, kClosed } state_;
 
  private:
   friend class Environment;
@@ -109,7 +110,6 @@ class HandleWrap : public AsyncWrap {
   // refer to `doc/contributing/node-postmortem-support.md`
   friend int GenDebugSymbols();
   ListNode<HandleWrap> handle_wrap_queue_;
-  enum { kInitialized, kClosing, kClosed } state_;
   uv_handle_t* const handle_;
 };
 

src/tcp_wrap.cc

@@ -97,6 +97,7 @@ void TCPWrap::Initialize(Local<Object> target,
                       GetSockOrPeerName<TCPWrap, uv_tcp_getpeername>);
   env->SetProtoMethod(t, "setNoDelay", SetNoDelay);
   env->SetProtoMethod(t, "setKeepAlive", SetKeepAlive);
+  env->SetProtoMethod(t, "reset", Reset);
 
 #ifdef _WIN32
   env->SetProtoMethod(t, "setSimultaneousAccepts", SetSimultaneousAccepts);
@@ -134,6 +135,7 @@ void TCPWrap::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(GetSockOrPeerName<TCPWrap, uv_tcp_getpeername>);
   registry->Register(SetNoDelay);
   registry->Register(SetKeepAlive);
+  registry->Register(Reset);
 #ifdef _WIN32
   registry->Register(SetSimultaneousAccepts);
 #endif
@@ -339,7 +341,29 @@ void TCPWrap::Connect(const FunctionCallbackInfo<Value>& args,
 
   args.GetReturnValue().Set(err);
 }
+void TCPWrap::Reset(const FunctionCallbackInfo<Value>& args) {
+  TCPWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(
+      &wrap, args.Holder(), args.GetReturnValue().Set(UV_EBADF));
+
+  int err = wrap->Reset(args[0]);
+
+  args.GetReturnValue().Set(err);
+}
+
+int TCPWrap::Reset(Local<Value> close_callback) {
+  if (state_ != kInitialized) return 0;
 
+  int err = uv_tcp_close_reset(&handle_, OnClose);
+  state_ = kClosing;
+  if (!err & !close_callback.IsEmpty() && close_callback->IsFunction() &&
+      !persistent().IsEmpty()) {
+    object()
+        ->Set(env()->context(), env()->handle_onclose_symbol(), close_callback)
+        .Check();
+  }
+  return err;
+}
 
 // also used by udp_wrap.cc
 MaybeLocal<Object> AddressToJS(Environment* env,

src/tcp_wrap.h

@@ -88,6 +88,8 @@ class TCPWrap : public ConnectionWrap<TCPWrap, uv_tcp_t> {
       const v8::FunctionCallbackInfo<v8::Value>& args,
       int family,
       std::function<int(const char* ip_address, int port, T* addr)> uv_ip_addr);
+  static void Reset(const v8::FunctionCallbackInfo<v8::Value>& args);
+  int Reset(v8::Local<v8::Value> close_callback = v8::Local<v8::Value>());
 
 #ifdef _WIN32
   static void SetSimultaneousAccepts(

test/parallel/test-net-connect-reset-after-destroy.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+const assert = require('assert');
+
+const server = net.createServer();
+server.listen(0, common.mustCall(function() {
+  const port = server.address().port;
+  const conn = net.createConnection(port);
+  server.on('connection', (socket) => {
+    socket.on('error', common.expectsError({
+      code: 'ECONNRESET',
+      message: 'read ECONNRESET',
+      name: 'Error'
+    }));
+  });
+
+  conn.on('connect', common.mustCall(function() {
+    assert.strictEqual(conn, conn.resetAndDestroy().destroy());
+    conn.on('error', common.mustNotCall());
+
+    conn.write(Buffer.from('fzfzfzfzfz'), common.expectsError({
+      code: 'ERR_STREAM_DESTROYED',
+      message: 'Cannot call write after a stream was destroyed',
+      name: 'Error'
+    }));
+    server.close();
+  }));
+}));

test/parallel/test-net-connect-reset-before-connected.js

@@ -0,0 +1,13 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+
+const server = net.createServer();
+server.listen(0);
+const port = server.address().port;
+const socket = net.connect(port, common.localhostIPv4, common.mustNotCall());
+socket.on('error', common.mustNotCall());
+server.close();
+socket.resetAndDestroy();
+// `reset` waiting socket connected to sent the RST packet
+socket.destroy();

test/parallel/test-net-connect-reset-until-connected.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+const net = require('net');
+
+const server = net.createServer();
+server.listen(0, common.mustCall(function() {
+  const port = server.address().port;
+  const conn = net.createConnection(port);
+  conn.on('close', common.mustCall());
+  server.on('connection', (socket) => {
+    socket.on('error', common.expectsError({
+      code: 'ECONNRESET',
+      message: 'read ECONNRESET',
+      name: 'Error'
+    }));
+    server.close();
+  });
+  conn.resetAndDestroy();
+}));

test/parallel/test-net-connect-reset.js

@@ -0,0 +1,13 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+
+const socket = new net.Socket();
+socket.resetAndDestroy();
+// Emit error if socket is not connecting/connected
+socket.on('error', common.mustCall(
+  common.expectsError({
+    code: 'ERR_SOCKET_CLOSED',
+    name: 'Error'
+  }))
+);

test/parallel/test-net-server-reset.js

@@ -0,0 +1,36 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const net = require('net');
+
+const sockets = [];
+
+const server = net.createServer(function(c) {
+  c.on('close', common.mustCall());
+
+  sockets.push(c);
+
+  if (sockets.length === 2) {
+    assert.strictEqual(server.close(), server);
+    sockets.forEach((c) => c.resetAndDestroy());
+  }
+});
+
+server.on('close', common.mustCall());
+
+assert.strictEqual(server, server.listen(0, () => {
+  net.createConnection(server.address().port)
+    .on('error', common.mustCall(
+      common.expectsError({
+        code: 'ECONNRESET',
+        name: 'Error'
+      }))
+    );
+  net.createConnection(server.address().port)
+    .on('error', common.mustCall(
+      common.expectsError({
+        code: 'ECONNRESET',
+        name: 'Error'
+      }))
+    );
+}));

test/parallel/test-net-socket-reset-send.js

@@ -0,0 +1,30 @@
+'use strict';
+
+const common = require('../common');
+const net = require('net');
+const assert = require('assert');
+
+const server = net.createServer();
+server.listen(0, common.mustCall(() => {
+  const port = server.address().port;
+  const conn = net.createConnection(port);
+  server.on('connection', (socket) => {
+    socket.on('error', common.expectsError({
+      code: 'ECONNRESET',
+      message: 'read ECONNRESET',
+      name: 'Error'
+    }));
+  });
+
+  conn.on('connect', common.mustCall(() => {
+    assert.strictEqual(conn, conn.resetAndDestroy().destroy());
+    conn.on('error', common.mustNotCall());
+
+    conn.write(Buffer.from('fzfzfzfzfz'), common.expectsError({
+      code: 'ERR_STREAM_DESTROYED',
+      message: 'Cannot call write after a stream was destroyed',
+      name: 'Error'
+    }));
+    server.close();
+  }));
+}));

test/parallel/test-net-socket-reset-twice.js

@@ -0,0 +1,15 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+
+const server = net.createServer();
+server.listen(0);
+const port = server.address().port;
+const conn = net.createConnection(port);
+
+conn.on('error', common.mustCall(() => {
+  conn.resetAndDestroy();
+}));
+
+conn.on('close', common.mustCall());
+server.close();