Change logic around closing connections and writers API-1283 (#1417)

The issue was writing some data to the connection after it being closed. I changed and refactored some logic around PipelinedWriter and Connection to avoid that. 

- Moved socket closing responsibility to Writers
- Fixed a test in PipelinedWriterTest
- Ongoing and future socket.write calls will end with an error after the writer is closed. This is because close() destroys the socket and any ongoing and future socket.write calls will end with an error as described in node.js [documentation](https://nodejs.org/api/stream.html#writabledestroyerror). So we will be able to reject all deferred promises in the write queue upon close due to connection.close() or socket write error close().
- We were not closing the socket on write error before, now we do it. We were sending end packet and don't write anymore. This is the same as java. [See](https://hazelcast.slack.com/archives/C01JU7ZJYGP/p1668695419424179)

fixes #1256

Author: srknzl

src/network/Connection.ts

@@ -41,7 +41,7 @@ abstract class Writer extends EventEmitter {
 
     abstract write(message: ClientMessage, resolver: DeferredPromise<void>): void;
 
-    abstract close(): void;
+    abstract close(cause: Error): void;
 
 }
 
@@ -81,15 +81,22 @@ export class PipelinedWriter extends Writer {
 
     write(message: ClientMessage, resolver: DeferredPromise<void>): void {
         if (this.error) {
-            // if there was a write error, it's useless to keep writing to the socket
+            // if the socket is closed, it's useless to keep writing to the socket
             return process.nextTick(() => resolver.reject(this.error));
         }
         this.queue.push({ message, resolver });
         this.schedule();
     }
 
-    close(): void {
+    close(error: Error): void {
+        if (this.error) {
+            return;
+        }
+        this.error = this.makeIOError(error);
         this.canWrite = false;
+        // If we pass an error to destroy, an unhandled error will be thrown because we don't handle the error event
+        // So we don't pass anything to the socket. It is internal anyway.
+        this.socket.destroy();
         // no more items can be added now
         this.queue = FROZEN_ARRAY;
     }
@@ -164,15 +171,22 @@ export class PipelinedWriter extends Writer {
         });
     }
 
-    private handleError(err: any, sentResolvers: OutputQueueItem[]): void {
-        this.error = new IOError(err);
+    private handleError(err: Error, sentResolvers: OutputQueueItem[]): void {
+        const error = this.makeIOError(err);
         for (const item of sentResolvers) {
-            item.resolver.reject(this.error);
+            item.resolver.reject(error);
         }
         for (const item of this.queue) {
-            item.resolver.reject(this.error);
+            item.resolver.reject(error);
+        }
+        this.close(error);
+    }
+
+    private makeIOError(err: Error): IOError {
+        if (err instanceof IOError) {
+            return err;
         }
-        this.close();
+        return new IOError(err.message, err);
     }
 }
 
@@ -199,8 +213,8 @@ export class DirectWriter extends Writer {
         });
     }
 
-    close(): void {
-        // no-op
+    close(cause: Error): void {
+        this.socket.destroy();
     }
 }
 
@@ -411,7 +425,7 @@ export class Connection {
     /**
      * Closes this connection.
      */
-    close(reason: string, cause: Error): void {
+    close(reason: string | null, cause: Error | null): void {
         if (this.closedTime !== 0) {
             return;
         }
@@ -422,8 +436,7 @@ export class Connection {
 
         this.logClose();
 
-        this.writer.close();
-        this.socket.end();
+        this.writer.close(this.closedCause ? this.closedCause : new Error(reason ? reason : 'Connection closed'));
 
         this.connectionManager.onConnectionClose(this);
     }

test/unit/connection/DirectWriterTest.js

@@ -26,9 +26,10 @@ const {
     Frame
 } = require('../../../lib/protocol/ClientMessage');
 const { deferredPromise } = require('../../../lib/util/Util');
+const sandbox = sinon.createSandbox();
 
 describe('DirectWriterTest', function () {
-    let queue;
+    let writer;
     let mockSocket;
     let writtenBytes;
 
@@ -40,25 +41,29 @@ describe('DirectWriterTest', function () {
 
     const setUpWriteSuccess = () => {
         mockSocket = new Socket({});
-        sinon.stub(mockSocket, 'write').callsFake((data, cb) => {
+        sandbox.stub(mockSocket, 'write').callsFake((data, cb) => {
             cb();
             mockSocket.emit('data', data);
         });
-        queue = new DirectWriter(mockSocket, numberOfBytes => {
+        writer = new DirectWriter(mockSocket, numberOfBytes => {
             writtenBytes += numberOfBytes;
         });
     };
 
     const setUpWriteFailure = (err) => {
         mockSocket = new Socket({});
-        sinon.stub(mockSocket, 'write').callsFake((_, cb) => {
+        sandbox.stub(mockSocket, 'write').callsFake((_, cb) => {
             cb(err);
         });
-        queue = new DirectWriter(mockSocket, numberOfBytes => {
+        writer = new DirectWriter(mockSocket, numberOfBytes => {
             writtenBytes += numberOfBytes;
         });
     };
 
+    afterEach(function() {
+        sandbox.restore();
+    });
+
     it('increment written bytes correctly', function(done) {
         setUpWriteSuccess();
 
@@ -70,7 +75,7 @@ describe('DirectWriterTest', function () {
             done();
         });
 
-        queue.write(msg, deferredPromise());
+        writer.write(msg, deferredPromise());
     });
 
     it('writes single message into socket', function(done) {
@@ -82,7 +87,7 @@ describe('DirectWriterTest', function () {
             done();
         });
 
-        queue.write(msg, deferredPromise());
+        writer.write(msg, deferredPromise());
     });
 
     it('writes multiple messages separately into socket', function(done) {
@@ -97,16 +102,16 @@ describe('DirectWriterTest', function () {
             }
         });
 
-        queue.write(msg, deferredPromise());
-        queue.write(msg, deferredPromise());
-        queue.write(msg, deferredPromise());
+        writer.write(msg, deferredPromise());
+        writer.write(msg, deferredPromise());
+        writer.write(msg, deferredPromise());
     });
 
     it('resolves promise on write success', function(done) {
         setUpWriteSuccess();
 
         const resolver = deferredPromise();
-        queue.write(createMessage('test'), resolver);
+        writer.write(createMessage('test'), resolver);
         resolver.promise.then(done);
     });
 
@@ -115,7 +120,7 @@ describe('DirectWriterTest', function () {
         setUpWriteFailure(err);
 
         const resolver = deferredPromise();
-        queue.write(createMessage('test'), resolver);
+        writer.write(createMessage('test'), resolver);
         resolver.promise.catch((err) => {
             expect(err).to.be.equal(err);
             done();
@@ -125,18 +130,29 @@ describe('DirectWriterTest', function () {
     it('emits write event on write success', function(done) {
         setUpWriteSuccess();
 
-        queue.on('write', done);
-        queue.write(createMessage('test'), deferredPromise());
+        writer.on('write', done);
+        writer.write(createMessage('test'), deferredPromise());
     });
 
     it('does not emit write event on write failure', function(done) {
         setUpWriteFailure(new Error());
 
-        queue.on('write', () => done(new Error()));
+        writer.on('write', () => done(new Error()));
         const resolver = deferredPromise();
-        queue.write(createMessage('test'), resolver);
+        writer.write(createMessage('test'), resolver);
         resolver.promise.catch(() => {
             done();
         });
     });
+
+    it('should close the socket upon being closed', function() {
+        setUpWriteSuccess();
+
+        // This is equivalent to a sinon spy
+        const spy = sandbox.fake(mockSocket.destroy);
+        sandbox.replace(mockSocket, 'destroy', spy);
+        writer.close();
+
+        expect(spy.calledOnce).to.be.true;
+    });
 });

test/unit/connection/PipelinedWriterTest.js

@@ -20,11 +20,14 @@ const sinon = require('sinon');
 const { expect } = require('chai');
 
 const { PipelinedWriter } = require('../../../lib/network/Connection');
+const { IOError } = require('../../../lib/core/HazelcastError');
+const TestUtil = require('../../TestUtil');
 const {
     ClientMessage,
     Frame
 } = require('../../../lib/protocol/ClientMessage');
 const { deferredPromise } = require('../../../lib/util/Util');
+const sandbox = sinon.createSandbox();
 
 describe('PipelinedWriterTest', function () {
     const THRESHOLD = 8192;
@@ -35,7 +38,7 @@ describe('PipelinedWriterTest', function () {
 
     function setUpWriteSuccess(canWrite) {
         mockSocket = new Socket({});
-        sinon.stub(mockSocket, 'write').callsFake((data, cb) => {
+        sandbox.stub(mockSocket, 'write').callsFake((data, cb) => {
             process.nextTick(cb);
             process.nextTick(() => mockSocket.emit('data', data));
             return canWrite;
@@ -47,7 +50,7 @@ describe('PipelinedWriterTest', function () {
 
     function setUpWriteFailure(err) {
         mockSocket = new Socket({});
-        sinon.stub(mockSocket, 'write').callsFake((_, cb) => {
+        sandbox.stub(mockSocket, 'write').callsFake((_, cb) => {
             process.nextTick(() => cb(err));
             return false;
         });
@@ -68,6 +71,10 @@ describe('PipelinedWriterTest', function () {
         return clientMessage;
     }
 
+    afterEach(function () {
+        sandbox.restore();
+    });
+
     it('increment written bytes correctly', function(done) {
         setUpWriteSuccess(true);
 
@@ -199,16 +206,17 @@ describe('PipelinedWriterTest', function () {
         Promise.all([resolver1.promise, resolver2.promise]).then(() => done());
     });
 
-    it('rejects single promise on write failure', function(done) {
+    it('rejects single promise on write failure', async function() {
         const err = new Error();
         setUpWriteFailure(err);
 
         const resolver = deferredPromise();
         writer.write(createMessageFromString('test'), resolver);
-        resolver.promise.catch((err) => {
-            expect(err).to.be.equal(err);
-            done();
+        const rejReason = await TestUtil.getRejectionReasonOrThrow(async () => {
+            await resolver.promise;
         });
+        expect(rejReason.cause).to.be.equal(err);
+        expect(rejReason).to.be.instanceOf(IOError);
     });
 
     it('rejects multiple promises on write failure', function(done) {
@@ -276,4 +284,39 @@ describe('PipelinedWriterTest', function () {
             writer.write(msg, deferredPromise());
         });
     });
+
+    it('should not schedule a write if a write() is called when the writer is already closed', async function() {
+        setUpWriteSuccess(true);
+
+        const msg = createMessageFromString('test');
+        // Pass a IOError so that the same error is used to reject the write() deferred promise
+        const closeReason = new IOError();
+        writer.close(closeReason);
+        const deferred = deferredPromise();
+
+        // This is equivalent to a sinon spy
+        const spy = sandbox.fake(writer.schedule);
+        sandbox.replace(writer, 'schedule', spy);
+        writer.write(msg, deferred);
+
+        const rejectionReason = await TestUtil.getRejectionReasonOrThrow(async () => await deferred.promise);
+        expect(rejectionReason).to.be.equal(closeReason);
+        expect(spy.callCount).to.be.equal(0);
+    });
+
+    it('should not destroy the socket twice upon closing again', async function() {
+        setUpWriteSuccess(true);
+
+        // Pass a IOError so that the same error is used to reject the write() deferred promise
+        const closeReason = new IOError();
+
+        // This is equivalent to a sinon spy
+        const spy = sandbox.fake(mockSocket.destroy);
+        sandbox.replace(mockSocket, 'destroy', spy);
+
+        writer.close(closeReason);
+        writer.close(closeReason);
+
+        expect(spy.callCount).to.be.equal(1);
+    });
 });