Implement a keep-alive timer for /sync requests

When a /sync request fails, we spin up a keep-alive poll to /_matrix/client/r0
which 400s. We treat any HTTP response code as a success for the purposes of
polling the server. When a successful poll is done, we shoot the current /sync
request in the head immediately (via a hacky abort() on the promise) and retry
the /sync.

lib/http-api.js

@@ -43,6 +43,8 @@ module.exports.PREFIX_V2_ALPHA = "/_matrix/client/v2_alpha";
  */
 module.exports.PREFIX_IDENTITY_V1 = "/_matrix/identity/api/v1";
 
+module.exports.PREFIX_R0 = "/_matrix/client/r0";
+
 /**
  * Construct a MatrixHttpApi.
  * @constructor
@@ -416,8 +418,10 @@ module.exports.MatrixHttpApi.prototype = {
             }, localTimeoutMs);
         }
 
+        var reqPromise = defer.promise;
+
         try {
-            this.opts.request(
+            var req = this.opts.request(
                 {
                     uri: uri,
                     method: method,
@@ -425,6 +429,7 @@ module.exports.MatrixHttpApi.prototype = {
                     qs: queryParams,
                     body: data,
                     json: true,
+                    timeout: localTimeoutMs,
                     _matrix_opts: this.opts
                 },
                 function(err, response, body) {
@@ -438,6 +443,9 @@ module.exports.MatrixHttpApi.prototype = {
                     handlerFn(err, response, body);
                 }
             );
+            // FIXME: This is EVIL, but I can't think of a better way to expose
+            // abort() operations on underlying HTTP requests :(
+            reqPromise.abort = req.abort.bind(req);
         }
         catch (ex) {
             defer.reject(ex);
@@ -445,7 +453,7 @@ module.exports.MatrixHttpApi.prototype = {
                 callback(ex);
             }
         }
-        return defer.promise;
+        return reqPromise;
     }
 };
 

lib/sync.js

@@ -66,6 +66,8 @@ function SyncApi(client, opts) {
     opts.pendingEventOrdering = opts.pendingEventOrdering || "chronological";
     this.opts = opts;
     this._peekRoomId = null;
+    this._syncConnectionLost = false;
+    this._currentSyncRequest = null;
 }
 
 /**
@@ -348,10 +350,13 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
     // normal timeout= plus buffer time
     var clientSideTimeoutMs = this.opts.pollTimeout + BUFFER_PERIOD_MS;
 
-    client._http.authedRequestWithPrefix(
+    this._currentSyncRequest = client._http.authedRequestWithPrefix(
         undefined, "GET", "/sync", qps, undefined, httpApi.PREFIX_V2_ALPHA,
         clientSideTimeoutMs
-    ).done(function(data) {
+    );
+
+    this._currentSyncRequest.done(function(data) {
+        self._syncConnectionLost = false;
         // data looks like:
         // {
         //    next_batch: $token,
@@ -525,6 +530,21 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
 
         self._sync(syncOptions);
     }, function(err) {
+        if (!self._syncConnectionLost) {
+            debuglog("Starting keep-alive");
+            self._syncConnectionLost = true;
+            retryPromise(self._pokeKeepAlive.bind(self), 2000).done(function() {
+                debuglog("Keep-alive successful.");
+                // blow away the current /sync request if the connection is still
+                // dead. It may be black-holed.
+                if (!self._syncConnectionLost) {
+                    return;
+                }
+                // kill the current sync request
+                debuglog("Aborting current /sync.");
+                self._currentSyncRequest.abort();
+            });
+        }
         console.error("/sync error (%s attempts): %s", attempt, err);
         console.error(err);
         attempt += 1;
@@ -535,6 +555,21 @@ SyncApi.prototype._sync = function(syncOptions, attempt) {
     });
 };
 
+/**
+ * @return {Promise}
+ */
+SyncApi.prototype._pokeKeepAlive = function() {
+    return this.client._http.requestWithPrefix(
+        undefined, "GET", "/", undefined,
+        undefined, httpApi.PREFIX_R0, 5 * 1000
+    ).catch(function(err) {
+        if (err.httpStatus > 0) { // we hit the server alright
+            return q();
+        }
+        throw err;
+    });
+};
+
 /**
  * @param {string} filterName
  * @param {Filter} filter
@@ -714,6 +749,14 @@ function retryTimeMsForAttempt(attempt) {
     return Math.pow(2, Math.min(attempt, 5)) * 1000;
 }
 
+function retryPromise(promiseFn, delay) {
+    delay = delay || 0;
+    return promiseFn().catch(function(reason) { // if it fails
+        // retry after waiting the delay time
+        return q.delay(delay).then(retryPromise.bind(null, promiseFn, delay));
+    });
+}
+
 function startSyncingRetryTimer(client, attempt, fn) {
     client._syncingRetry = {};
     client._syncingRetry.fn = fn;