[API-424] Do not allow concurrent requests on SessionManager for the same group id

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);
+            }
+
+            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());