Merge pull request #371 from uProxy/trevj-retry

retry for SSH connections, with a re-useable library

Author: trevj

src/churn-pipe/churn-pipe.ts

@@ -12,30 +12,14 @@ import ipaddr = require('ipaddr.js');
 import logging = require('../logging/logging');
 import net = require('../net/net.types');
 import PassThrough = require('../transformers/passthrough');
+import promises = require('../promises/promises');
 import protean = require('../transformers/protean');
 import sequence = require('../transformers/byteSequenceShaper');
 
 import Socket = freedom.UdpSocket.Socket;
 
 var log :logging.Log = new logging.Log('churn-pipe');
 
-// Retry an async function with exponential backoff for up to 2 seconds
-// before failing.
-var retry_ = <T>(func:() => Promise<T>, delayMs?:number) : Promise<T> => {
-  delayMs = delayMs || 10;
-  return func().catch((err) => {
-    delayMs *= 2;
-    if (delayMs > 2000) {
-      return Promise.reject(err);
-    }
-    return new Promise<T>((F, R) => {
-      setTimeout(() => {
-        retry_(func, delayMs).then(F, R);
-      }, delayMs);
-    });
-  });
-}
-
 // Maps transformer names to class constructors.
 var transformers :{[name:string] : new() => Transformer} = {
   'caesar': caesar.CaesarCipher,
@@ -47,6 +31,10 @@ var transformers :{[name:string] : new() => Transformer} = {
   'sequenceShaper': sequence.ByteSequenceShaper
 };
 
+// Local socket rebinding retry timing (see bindLocal)
+const INITIAL_REBIND_INTERVAL_MS = 10;
+const MAX_REBIND_INTERVAL_MS = 2000;
+
 interface MirrorSet {
   // If true, these mirrors represent a remote endpoint that has been
   // explicitly signaled to us.
@@ -198,13 +186,13 @@ class Pipe {
     // This retry is needed because the browser releases the UDP port
     // asynchronously after we call close() on the RTCPeerConnection, so
     // this call to bind() may initially fail, until the port is released.
-    portPromise = retry_(() => {
+    portPromise = promises.retryWithExponentialBackoff(() => {
       log.debug('%1: trying to bind public endpoint: %2',
           this.name_, publicEndpoint);
       // TODO: Once https://github.com/freedomjs/freedom/issues/283 is
       // fixed, catch here, and only retry on an ALREADY_BOUND error.
       return socket.bind(anyInterface, publicEndpoint.port);
-    }).then(() => {
+    }, MAX_REBIND_INTERVAL_MS, INITIAL_REBIND_INTERVAL_MS).then(() => {
       log.debug('%1: successfully bound public endpoint: %2',
           this.name_, publicEndpoint);
       socket.on('onData', (recvFromInfo:freedom.UdpSocket.RecvFromInfo) => {

src/cloud/social/provider.ts

@@ -6,6 +6,7 @@
 import arraybuffers = require('../../arraybuffers/arraybuffers');
 import linefeeder = require('../../net/linefeeder');
 import logging = require('../../logging/logging');
+import promises = require('../../promises/promises');
 import queue = require('../../handler/queue');
 
 // https://github.com/borisyankov/DefinitelyTyped/blob/master/ssh2/ssh2-tests.ts
@@ -25,6 +26,9 @@ const STORAGE_KEY = 'cloud-social-contacts';
 // Timeout for establishing an SSH connection.
 const CONNECT_TIMEOUT_MS = 10000;
 
+// Maximum number of times to try establishing an SSH connection.
+const MAX_CONNECT_ATTEMPTS = 3;
+
 // Credentials for accessing a cloud instance.
 // The serialised, base64 form is distributed amongst users.
 interface Invite {
@@ -199,15 +203,23 @@ export class CloudSocialProvider {
       });
     }
 
-    var connection = new Connection(invite, (message: Object) => {
-      this.dispatchEvent_('onMessage', {
-        from: makeClientState(invite.host),
-        // SIGNAL_FROM_SERVER_PEER,
-        message: JSON.stringify(makeVersionedPeerMessage(3002, message))
+    let numAttempts = 0;
+    const connect = () => {
+      log.debug('connection attempt %1...', (++numAttempts));
+      const connection = new Connection(invite, (message: Object) => {
+        this.dispatchEvent_('onMessage', {
+          from: makeClientState(invite.host),
+          // SIGNAL_FROM_SERVER_PEER,
+          message: JSON.stringify(makeVersionedPeerMessage(3002, message))
+        });
       });
-    });
-
-    this.clients_[invite.host] = connection.connect().then(() => {
+      return connection.connect().then(() => {
+        return connection;
+      });
+    };
+        
+    this.clients_[invite.host] = promises.retry(connect,
+        MAX_CONNECT_ATTEMPTS).then((connection:Connection) => {
       log.info('connected to zork on %1', invite.host);
 
       // Fetch the banner, if available, then emit an instance message.

src/promises/promises.spec.ts

@@ -0,0 +1,49 @@
+/// <reference path='../../../third_party/typings/es6-promise/es6-promise.d.ts' />
+/// <reference path='../../../third_party/typings/jasmine/jasmine.d.ts' />
+
+import promises = require('./promises');
+
+describe('retry', () => {
+  const MAX_ATTEMPTS = 5;
+  const RETURN_VALUE = 'hello';
+  const REJECT_MESSAGE = 'goodbye';
+
+  it('only calls successful function once', (done) => {
+    const f = jasmine.createSpy('spy');
+    f.and.returnValue(Promise.resolve(RETURN_VALUE));
+    promises.retry(f, MAX_ATTEMPTS).then((result: string) => {
+      expect(f.calls.count()).toEqual(1);
+      expect(result).toEqual(RETURN_VALUE);
+      done();
+    });
+  });
+
+  it('calls multiple times until success', (done) => {
+    const NUM_CALLS_BEFORE_SUCCESS = 3;
+    let callCount = 0;
+    const f = jasmine.createSpy('spy');
+    f.and.callFake(() => {
+      callCount++;
+      if (callCount === NUM_CALLS_BEFORE_SUCCESS) {
+        return Promise.resolve(RETURN_VALUE);
+      } else {
+        return Promise.reject('error');
+      }
+    });
+    promises.retry(f, NUM_CALLS_BEFORE_SUCCESS + 1).then((result: string) => {
+      expect(f.calls.count()).toEqual(NUM_CALLS_BEFORE_SUCCESS);
+      expect(result).toEqual(RETURN_VALUE);
+      done();
+    });
+  });
+
+  it('stops calling after the max number of failures', (done) => {
+    const f = jasmine.createSpy('spy');
+    f.and.returnValue(Promise.reject(new Error(REJECT_MESSAGE)));
+    promises.retry(f, MAX_ATTEMPTS).catch((e: Error) => {
+      expect(f.calls.count()).toEqual(MAX_ATTEMPTS);
+      expect(e.message).toEqual(REJECT_MESSAGE);
+      done();
+    });
+  });
+});

src/promises/promises.ts

@@ -0,0 +1,31 @@
+/// <reference path='../../../third_party/typings/es6-promise/es6-promise.d.ts' />
+
+// Invokes f up to maxAttempts number of times, resolving with its result
+// on the first success and rejecting on maxAttempts-th failure.
+export const retry = <T>(f: () => Promise<T>, maxAttempts: number): Promise<T> => {
+  return f().catch((e:Error) => {
+    --maxAttempts;
+    if (maxAttempts > 0) {
+      return retry(f, maxAttempts);
+    } else {
+      return Promise.reject(e);
+    }
+  });
+};
+
+// Invokes f with exponential backoff between retries, resolving with its result
+// on the first success and rejecting on maxAttempts-th failure.
+export const retryWithExponentialBackoff = <T>(f: () => Promise<T>,
+    maxIntervalMs: number, initialIntervalMs: number): Promise<T> => {
+  return f().catch((e: Error) => {
+    initialIntervalMs *= 2;
+    if (initialIntervalMs > maxIntervalMs) {
+      return Promise.reject(e);
+    }
+    return new Promise<T>((F, R) => {
+      setTimeout(() => {
+        retryWithExponentialBackoff(f, maxIntervalMs, initialIntervalMs).then(F, R);
+      }, initialIntervalMs);
+    });
+  });
+};