[API-424] Do not allow concurrent requests on SessionManager for the …

…same group id (#904)

* Do not allow concurrent requests on SessionManager for the same group id

We should make sure that for the same group id, new create session requests
should only be made if there is no session or the session is expired.

To achieve this, a new Map that holds the group id to in-flight
create session requests pairs is introduced. When an in-flight request
is found for a group id, concurrent requests now wait for the initial
request to complete (either normally or exceptionally) and then try
again.

* check the error text in the test

src/proxy/cpsubsystem/CPSessionManager.ts

@@ -87,6 +87,8 @@ export class CPSessionManager {
 
     // <group_id, session_state> map
     private readonly sessions: Map<string, SessionState> = new Map();
+    // group id to in-flight create session requests map
+    private readonly inFlightCreateSessionRequests: Map<string, Promise<SessionState>> = new Map();
     private heartbeatTask: Task;
     private isShutdown = false;
 
@@ -142,7 +144,10 @@ export class CPSessionManager {
             .catch((e) => {
                 this.logger.debug('CPSessionManager', 'Could not close CP sessions.', e);
             })
-            .then(() => this.sessions.clear());
+            .then(() => {
+                this.sessions.clear();
+                this.inFlightCreateSessionRequests.clear();
+            });
     }
 
     private getOrCreateSession(groupId: RaftGroupId): Promise<SessionState> {
@@ -157,21 +162,38 @@ export class CPSessionManager {
     }
 
     private createNewSession(groupId: RaftGroupId): Promise<SessionState> {
-        return this.requestNewSession(groupId).then((response) => {
-            const state = new SessionState(response.sessionId, groupId, response.ttlMillis.toNumber());
-            this.sessions.set(groupId.getStringId(), state);
-            this.scheduleHeartbeatTask(response.heartbeatMillis.toNumber());
-            return state;
-        });
+        // Check if there is a session request for this group id in-flight.
+        const inFlightRequest = this.inFlightCreateSessionRequests.get(groupId.getStringId());
+        if (inFlightRequest === undefined) {
+            // No in-flight request for this group id. Let's make a new one and register it.
+            const requestNewSessionPromise = this.requestNewSession(groupId).then((response) => {
+                const state = new SessionState(response.sessionId, groupId, response.ttlMillis.toNumber());
+                this.sessions.set(groupId.getStringId(), state);
+                this.scheduleHeartbeatTask(response.heartbeatMillis.toNumber());
+                return state;
+            });
+            this.inFlightCreateSessionRequests.set(groupId.getStringId(), requestNewSessionPromise);
+
+            // Remove the request once it is completed normally or exceptionally.
+            // This ensures that, if there is no session(request failed or session somehow removed later)
+            // or the session is expired, later, a new request can be made for the same group id.
+            const onResponseOrError = () => this.inFlightCreateSessionRequests.delete(groupId.getStringId());
+            requestNewSessionPromise.then(onResponseOrError, onResponseOrError);
+            return requestNewSessionPromise;
+        } else {
+            // There is an in-flight session request for this group id. We should
+            // wait for it to complete normally or exceptionally, and then try again.
+            // There should be no concurrent create session requests for the same
+            // group id.
+            const onResponseOrError = () => this.getOrCreateSession(groupId);
+            return inFlightRequest.then(onResponseOrError, onResponseOrError);
+        }
     }
 
     private requestNewSession(groupId: RaftGroupId): Promise<CPSessionCreateSessionResponseParams> {
         const clientMessage = CPSessionCreateSessionCodec.encodeRequest(groupId, this.clientName);
         return this.invocationService.invokeOnRandomTarget(clientMessage)
-            .then((clientMessage) => {
-                const response = CPSessionCreateSessionCodec.decodeResponse(clientMessage);
-                return response;
-            });
+            .then(CPSessionCreateSessionCodec.decodeResponse);
     }
 
     private requestCloseSession(groupId: RaftGroupId, sessionId: Long): Promise<boolean> {

test/unit/proxy/cpsubsystem/CPSessionManagerTest.js

@@ -32,6 +32,7 @@ const {
 } = require('../../../../lib/proxy/cpsubsystem/CPSessionManager');
 const { DefaultLogger } = require('../../../../lib/logging/DefaultLogger');
 const { RaftGroupId } = require('../../../../lib/proxy/cpsubsystem/RaftGroupId');
+const { deferredPromise } = require('../../../../lib/util/Util');
 
 describe('CPSessionManagerTest', function () {
 
@@ -114,13 +115,17 @@ describe('CPSessionManagerTest', function () {
             return new SessionState(Long.fromNumber(SESSION_ID), groupId, TTL_MILLIS);
         }
 
-        function stubRequestNewSession() {
-            const stub = sandbox.stub(sessionManager, 'requestNewSession');
-            stub.returns(Promise.resolve({
+        function prepareNewSessionResponse() {
+            return {
                 sessionId: Long.fromNumber(SESSION_ID),
                 ttlMillis: Long.fromNumber(TTL_MILLIS),
                 heartbeatMillis: Long.fromNumber(HEARTBEAT_MILLIS)
-            }));
+            };
+        }
+
+        function stubRequestNewSession() {
+            const stub = sandbox.stub(sessionManager, 'requestNewSession');
+            stub.returns(Promise.resolve(prepareNewSessionResponse()));
             return stub;
         }
 
@@ -200,6 +205,53 @@ describe('CPSessionManagerTest', function () {
             expect(sessionManager.sessions.get(GROUP_ID_AS_STRING).acquireCount).to.be.equal(1);
         });
 
+        it('acquireSession: should not request new session for the same group id for concurrent requests', async function () {
+            const stub = sandbox.stub(sessionManager, 'requestNewSession');
+            const deferred = deferredPromise();
+            stub.returns(deferred.promise);
+
+            const groupId = prepareGroupId();
+
+            const acquireSessionPromises = [
+                sessionManager.acquireSession(groupId),
+                sessionManager.acquireSession(groupId),
+                sessionManager.acquireSession(groupId),
+            ];
+
+            expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(1);
+
+            deferred.resolve(prepareNewSessionResponse());
+
+            await Promise.all(acquireSessionPromises);
+            expect(stub.withArgs(groupId).calledOnce).to.be.true;
+            expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(0);
+        });
+
+        it('acquireSession: should request new sessions for concurrent requests when requests fail', async function () {
+            const stub = sandbox.stub(sessionManager, 'requestNewSession');
+            const deferred = deferredPromise();
+            stub.returns(deferred.promise);
+
+            const groupId = prepareGroupId();
+
+            const acquireSessionPromises = [
+                sessionManager.acquireSession(groupId),
+                sessionManager.acquireSession(groupId),
+                sessionManager.acquireSession(groupId),
+            ];
+
+            expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(1);
+
+            deferred.resolve(Promise.reject(new Error('expected')));
+
+            for (const acquireSessionPromise of acquireSessionPromises) {
+                await expect(acquireSessionPromise).to.be.rejectedWith(Error, 'expected');
+            }
+
+            expect(stub.withArgs(groupId).callCount).to.be.equal(acquireSessionPromises.length);
+            expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(0);
+        });
+
         it('releaseSession: should decrement acquire counter by 1 for known session', function () {
             sessionManager.sessions.set(GROUP_ID_AS_STRING, prepareSessionState());