On upgrade failure, pass along connected error and reset connection serial

Resetting the message serial is a bit of a hack - fixes the problem for
now but may change in the future depending on the conclusion of
ably/realtime#587

Author: SimonWoolf

Diff for: common/lib/transport/connectionmanager.js

@@ -286,10 +286,10 @@ var ConnectionManager = (function() {
 				/*  if ws and xhrs are connecting in parallel, delay xhrs activation to let ws go ahead */
 				if(transport.shortName !== optimalTransport && Utils.arrIn(self.getUpgradePossibilities(), optimalTransport)) {
 					setTimeout(function() {
-						self.scheduleTransportActivation(transport, connectionKey);
+						self.scheduleTransportActivation(error, transport, connectionKey, connectionSerial, connectionId, clientId);
 					}, self.options.timeouts.parallelUpgradeDelay);
 				} else {
-					self.scheduleTransportActivation(transport, connectionKey);
+					self.scheduleTransportActivation(error, transport, connectionKey, connectionSerial, connectionId, clientId);
 				}
 			} else {
 				self.activateTransport(error, transport, connectionKey, connectionSerial, connectionId, clientId);
@@ -324,7 +324,7 @@ var ConnectionManager = (function() {
 	 * @param transport, the transport instance
 	 * @param connectionKey
 	 */
-	ConnectionManager.prototype.scheduleTransportActivation = function(transport, connectionKey) {
+	ConnectionManager.prototype.scheduleTransportActivation = function(error, transport, connectionKey, connectionSerial, connectionId, clientId) {
 		if(this.state.state !== 'connected') {
 			/* This is most likely to happen for the delayed xhrs, when xhrs and ws are scheduled in parallel*/
 			Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'Current connection state (' + this.state.state + ') is not valid to upgrade in; abandoning upgrade');
@@ -355,17 +355,28 @@ var ConnectionManager = (function() {
 			if(self.state === self.states.connected)
 				self.state = self.states.synchronizing;
 
+			/* If the connectionId has changed, the upgrade hasn't worked. But as
+			* it's still an upgrade, realtime still expects a sync - it just needs to
+			* be a sync with the new connectionSerial (which will be -1). (And it
+			* needs to be set in the library, which is done by activateTransport). */
+			var connectionReset = connectionId !== self.connectionId,
+				newConnectionSerial = connectionReset ? connectionSerial : self.connectionSerial;
+
+			if(connectionReset) {
+				Logger.logAction(Logger.LOG_ERROR, 'ConnectionManager.scheduleTransportActivation()', 'Upgrade resulted in new connectionId; resetting library connectionSerial from ' + self.connectionSerial + ' to ' + newConnectionSerial + '; upgrade error was ' + error);
+			}
+
 			/* make this the active transport */
 			Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'Activating transport; transport = ' + transport);
 			/* if activateTransport returns that it has not done anything (eg because the connection is closing), don't bother syncing */
-			if(self.activateTransport(null, transport, connectionKey, self.connectionSerial, self.connectionId)) {
+			if(self.activateTransport(error, transport, connectionKey, newConnectionSerial, connectionId, clientId)) {
 				Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'Syncing transport; transport = ' + transport);
-				self.sync(transport, function(err, connectionSerial, connectionId) {
+				self.sync(transport, function(err, newConnectionSerial, connectionId) {
 					if(err) {
 						Logger.logAction(Logger.LOG_ERROR, 'ConnectionManager.scheduleTransportActivation()', 'Unexpected error attempting to sync transport; transport = ' + transport + '; err = ' + err);
 						return;
 					}
-					Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'sync successful upgraded transport; transport = ' + transport + '; connectionSerial = ' + connectionSerial + '; connectionId = ' + connectionId);
+					Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'sync successful on upgraded transport; transport = ' + transport + '; connectionSerial = ' + newConnectionSerial + '; connectionId = ' + connectionId);
 
 					Logger.logAction(Logger.LOG_MINOR, 'ConnectionManager.scheduleTransportActivation()', 'Sending queued messages on upgraded transport; transport = ' + transport);
 					/* Restore pre-sync state. If state has changed in the meantime,

Diff for: spec/realtime/upgrade.test.js

@@ -2,6 +2,7 @@
 
 define(['ably', 'shared_helper'], function(Ably, helper) {
 	var exports = {},
+		_exports = {},
 		rest,
 		publishAtIntervals = function(numMessages, channel, dataFn, onPublish){
 			for(var i = numMessages; i > 0; i--) {
@@ -492,5 +493,45 @@ define(['ably', 'shared_helper'], function(Ably, helper) {
 		}
 	};
 
+	exports.unrecoverableUpgrade = function(test) {
+		test.expect(6);
+		var realtime,
+			fakeConnectionKey = '_____!ablyjs_test_fake-key____',
+			fakeConnectionId = 'ablyjs_tes';
+
+		try {
+			/* on base transport active */
+			realtime = helper.AblyRealtime();
+			realtime.connection.connectionManager.once('transport.active', function(transport) {
+				test.ok(transport.toString().indexOf('/comet/') > -1, 'assert first transport to become active is a comet transport');
+				test.equal(realtime.connection.errorReason, null, 'Check connection.errorReason is initially null');
+				/* sabotage the upgrade */
+				realtime.connection.connectionManager.connectionKey = fakeConnectionKey;
+				realtime.connection.connectionManager.connectionId = fakeConnectionId;
+
+				/* on upgrade failure */
+				realtime.connection.once('error', function(error) {
+					test.equal(error.code, 80008, 'Check correct (unrecoverable connection) error');
+					test.equal(realtime.connection.errorReason.code, 80008, 'Check error set in connection.errorReason');
+					test.equal(realtime.connection.state, 'connected', 'Check still connected');
+
+					/* Check events not still paused */
+					var channel = realtime.channels.get('unrecoverableUpgrade');
+					channel.attach(function(err) {
+						if(err) { test.ok(false, 'Attach error ' + helper.displayError(err)); }
+						channel.subscribe(function(msg) {
+							test.ok(true, 'Successfully received message');
+							closeAndFinish(test, realtime);
+						});
+						channel.publish('msg', null);
+					});
+				});
+			});
+		} catch(e) {
+			test.ok(false, 'test failed with exception: ' + e.stack);
+			closeAndFinish(test, realtime);
+		}
+	};
+
 	return module.exports = (bestTransport === 'web_socket') ? helper.withTimeout(exports) : {};
 });