Namespace Aphlict clients by request path, plus other fixes

Summary:
Fixes T7130. Fixes T7041. Fixes T7012.

Major change here is partitioning clients. In the Phacility cluster, being able to get a huge pile of instances on a single server -- without needing to run a process per instance -- is desirable.

To accomplish this, just bucket clients by the path they connect with. This will let us set client URIs to `/instancename/` and then route connections to a small set of servers. This degrades cleanly in the common case and has no effect on installs which don't do instancing.

Also fix two unrelated issues:

  - Fix the timeouts, which were incorrectly initializing in `open()` (which is called during reconnect, causing them to reset every time). Instead, initialize in the constructor. Cap timeout at 5 minutes.
  - Probably fix subscriptions, which were using a property with an object definition. Since this is by-ref, all concrete instances of the object share the same property, so all users would be subscribed to everything. Probably.

Test Plan:
  - Hit notification status page, saw version bump and instance/path name.
  - Saw instance/path name in client and server logs.
  - Stopped server, saw reconnects after 2, 4, 16, ... seconds.
  - Sent test notification; received test notification.
  - Didn't explicitly test the subscription thing but it should be obvious by looking at `/notification/status/` shortly after a push.

Reviewers: joshuaspence, btrahan

Reviewed By: btrahan

Subscribers: epriestley

Maniphest Tasks: T7041, T7012, T7130

Differential Revision: https://secure.phabricator.com/D11769

resources/celerity/map.php

@@ -7,8 +7,8 @@
  */
 return array(
   'names' => array(
-    'core.pkg.css' => 'f8f4b8dc',
+    'core.pkg.css' => '86353aff',
-    'core.pkg.js' => '65e04767',
+    'core.pkg.js' => '23d653bb',
     'darkconsole.pkg.js' => '8ab24e01',
     'differential.pkg.css' => '380f07e5',
     'differential.pkg.js' => '7b5a4aa4',
@@ -124,9 +124,9 @@ return array(
     'rsrc/css/phui/phui-action-list.css' => '9ee9910a',
     'rsrc/css/phui/phui-action-panel.css' => '4bcb288d',
     'rsrc/css/phui/phui-box.css' => '7b3a2eed',
-    'rsrc/css/phui/phui-button.css' => 'ffe12633',
+    'rsrc/css/phui/phui-button.css' => '008ba5e2',
     'rsrc/css/phui/phui-crumbs-view.css' => '594d719e',
-    'rsrc/css/phui/phui-document.css' => '8240b0b1',
+    'rsrc/css/phui/phui-document.css' => 'a494bdf8',
     'rsrc/css/phui/phui-error-view.css' => 'ad042fdd',
     'rsrc/css/phui/phui-feed-story.css' => 'c9f3a0b5',
     'rsrc/css/phui/phui-fontkit.css' => '9ae12677',
@@ -210,7 +210,7 @@ return array(
     'rsrc/externals/javelin/lib/Scrollbar.js' => '5b2f5a08',
     'rsrc/externals/javelin/lib/URI.js' => '6eff08aa',
     'rsrc/externals/javelin/lib/Vector.js' => '2caa8fb8',
-    'rsrc/externals/javelin/lib/WebSocket.js' => '3f840822',
+    'rsrc/externals/javelin/lib/WebSocket.js' => 'e292eaf4',
     'rsrc/externals/javelin/lib/Workflow.js' => '84d6aea0',
     'rsrc/externals/javelin/lib/__tests__/Cookie.js' => '5ed109e8',
     'rsrc/externals/javelin/lib/__tests__/DOM.js' => 'c984504b',
@@ -697,7 +697,7 @@ return array(
     'javelin-view-interpreter' => 'f829edb3',
     'javelin-view-renderer' => '6c2b09a2',
     'javelin-view-visitor' => 'efe49472',
-    'javelin-websocket' => '3f840822',
+    'javelin-websocket' => 'e292eaf4',
     'javelin-workflow' => '84d6aea0',
     'lightbox-attachment-css' => '7acac05d',
     'maniphest-batch-editor' => '8f380ebc',
@@ -773,13 +773,13 @@ return array(
     'phui-action-header-view-css' => '89c497e7',
     'phui-action-panel-css' => '4bcb288d',
     'phui-box-css' => '7b3a2eed',
-    'phui-button-css' => 'ffe12633',
+    'phui-button-css' => '008ba5e2',
     'phui-calendar-css' => '8675968e',
     'phui-calendar-day-css' => 'de035c8a',
     'phui-calendar-list-css' => 'c1d0ca59',
     'phui-calendar-month-css' => 'a92e47d2',
     'phui-crumbs-view-css' => '594d719e',
-    'phui-document-view-css' => '8240b0b1',
+    'phui-document-view-css' => 'a494bdf8',
     'phui-error-view-css' => 'ad042fdd',
     'phui-feed-story-css' => 'c9f3a0b5',
     'phui-font-icon-base-css' => '3dad2ae3',
@@ -1064,9 +1064,6 @@ return array(
       'javelin-dom',
       'phortune-credit-card-form',
     ),
-    '3f840822' => array(
-      'javelin-install',
-    ),
     '40a6a403' => array(
       'javelin-install',
       'javelin-dom',
@@ -1807,6 +1804,9 @@ return array(
       'javelin-stratcom',
       'javelin-dom',
     ),
+    'e292eaf4' => array(
+      'javelin-install',
+    ),
     'e32d14ab' => array(
       'javelin-behavior',
       'javelin-stratcom',

src/applications/notification/client/PhabricatorNotificationClient.php

@@ -2,12 +2,13 @@
 
 final class PhabricatorNotificationClient {
 
-  const EXPECT_VERSION = 6;
+  const EXPECT_VERSION = 7;
 
   public static function getServerStatus() {
     $uri = PhabricatorEnv::getEnvConfig('notification.server-uri');
     $uri = id(new PhutilURI($uri))
-      ->setPath('/status/');
+      ->setPath('/status/')
+      ->setQueryParam('instance', self::getInstance());
 
     list($body) = id(new HTTPSFuture($uri))
       ->setTimeout(3)
@@ -40,7 +41,8 @@ final class PhabricatorNotificationClient {
   private static function postMessage(array $data) {
     $server_uri = PhabricatorEnv::getEnvConfig('notification.server-uri');
     $server_uri = id(new PhutilURI($server_uri))
-      ->setPath('/');
+      ->setPath('/')
+      ->setQueryParam('instance', self::getInstance());
 
     id(new HTTPSFuture($server_uri, json_encode($data)))
       ->setMethod('POST')
@@ -48,4 +50,9 @@ final class PhabricatorNotificationClient {
       ->resolvex();
   }
 
+  private static function getInstance() {
+    $client_uri = PhabricatorEnv::getEnvConfig('notification.client-uri');
+    return id(new PhutilURI($client_uri))->getPath();
+  }
+
 }

src/applications/notification/controller/PhabricatorNotificationStatusController.php

@@ -44,6 +44,7 @@ final class PhabricatorNotificationStatusController
           $value = phutil_format_relative_time_detailed($value);
           break;
         case 'log':
+        case 'instance':
           break;
         default:
           $value = number_format($value);

support/aphlict/server/lib/AphlictAdminServer.js

@@ -5,6 +5,7 @@ var JX = require('./javelin').JX;
 require('./AphlictListenerList');
 
 var http = require('http');
+var url = require('url');
 
 JX.install('AphlictAdminServer', {
 
@@ -25,12 +26,8 @@ JX.install('AphlictAdminServer', {
     _server: null,
     _startTime: null,
 
-    getListeners: function() {
+    getListenerList: function(instance) {
-      return this.getListenerList().getListeners();
+      return this.getClientServer().getListenerList(instance);
-    },
-
-    getListenerList: function() {
-      return this.getClientServer().getListenerList();
     },
 
     listen: function() {
@@ -39,9 +36,11 @@ JX.install('AphlictAdminServer', {
 
     _handler: function(request, response) {
       var self = this;
+      var u = url.parse(request.url, true);
+      var instance = u.query.instance || '/';
 
       // Publishing a notification.
-      if (request.url == '/') {
+      if (u.pathname == '/') {
         if (request.method == 'POST') {
           var body = '';
 
@@ -54,11 +53,12 @@ JX.install('AphlictAdminServer', {
               var msg = JSON.parse(body);
 
               self.getLogger().log(
-                'Received notification: ' + JSON.stringify(msg));
+                'Received notification (' + instance + '): ' +
+                JSON.stringify(msg));
               ++self._messagesIn;
 
               try {
-                self._transmit(msg);
+                self._transmit(instance, msg);
                 response.writeHead(200, {'Content-Type': 'text/plain'});
               } catch (err) {
                 self.getLogger().log(
@@ -81,14 +81,17 @@ JX.install('AphlictAdminServer', {
           response.writeHead(405, 'Method Not Allowed');
           response.end();
         }
-      } else if (request.url == '/status/') {
+      } else if (u.pathname == '/status/') {
         var status = {
+          'instance': instance,
           'uptime': (new Date().getTime() - this._startTime),
-          'clients.active': this.getListenerList().getActiveListenerCount(),
+          'clients.active': this.getListenerList(instance)
-          'clients.total': this.getListenerList().getTotalListenerCount(),
+            .getActiveListenerCount(),
+          'clients.total': this.getListenerList(instance)
+            .getTotalListenerCount(),
           'messages.in': this._messagesIn,
           'messages.out': this._messagesOut,
-          'version': 6
+          'version': 7
         };
 
         response.writeHead(200, {'Content-Type': 'application/json'});
@@ -103,10 +106,12 @@ JX.install('AphlictAdminServer', {
     /**
      * Transmits a message to all subscribed listeners.
      */
-    _transmit: function(message) {
+    _transmit: function(instance, message) {
-      var listeners = this.getListeners().filter(function(client) {
+      var listeners = this.getListenerList(instance)
-        return client.isSubscribedToAny(message.subscribers);
+        .getListeners()
-      });
+        .filter(function(client) {
+          return client.isSubscribedToAny(message.subscribers);
+        });
 
       for (var i = 0; i < listeners.length; i++) {
         var listener = listeners[i];
@@ -119,7 +124,7 @@ JX.install('AphlictAdminServer', {
             '<%s> Wrote Message',
             listener.getDescription());
         } catch (error) {
-          this.getListenerList().removeListener(listener);
+          this.getListenerList(instance).removeListener(listener);
           this.getLogger().log(
             '<%s> Write Error: %s',
             listener.getDescription(),

support/aphlict/server/lib/AphlictClientServer.js

@@ -5,19 +5,28 @@ var JX = require('./javelin').JX;
 require('./AphlictListenerList');
 require('./AphlictLog');
 
+var url = require('url');
 var util = require('util');
 var WebSocket = require('ws');
 
 JX.install('AphlictClientServer', {
 
   construct: function(server) {
-    this.setListenerList(new JX.AphlictListenerList());
     this.setLogger(new JX.AphlictLog());
     this._server = server;
+    this._lists = {};
   },
 
   members: {
     _server: null,
+    _lists: null,
+
+    getListenerList: function(path) {
+      if (!this._lists[path]) {
+        this._lists[path] = new JX.AphlictListenerList(path);
+      }
+      return this._lists[path];
+    },
 
     listen: function() {
       var self = this;
@@ -25,7 +34,8 @@ JX.install('AphlictClientServer', {
       var wss = new WebSocket.Server({server: server});
 
       wss.on('connection', function(ws) {
-        var listener = self.getListenerList().addListener(ws);
+        var path = url.parse(ws.upgradeReq.url).pathname;
+        var listener = self.getListenerList(path).addListener(ws);
 
         function log() {
           self.getLogger().log(
@@ -70,12 +80,12 @@ JX.install('AphlictClientServer', {
         });
 
         ws.on('close', function() {
-          self.getListenerList().removeListener(listener);
+          self.getListenerList(path).removeListener(listener);
           log('Disconnected.');
         });
 
         wss.on('close', function() {
-          self.getListenerList().removeListener(listener);
+          self.getListenerList(path).removeListener(listener);
           log('Disconnected.');
         });
 
@@ -90,7 +100,6 @@ JX.install('AphlictClientServer', {
   },
 
   properties: {
-    listenerList: null,
     logger: null,
   }
 

support/aphlict/server/lib/AphlictListener.js

@@ -3,15 +3,18 @@
 var JX = require('./javelin').JX;
 
 JX.install('AphlictListener', {
-  construct: function(id, socket) {
+  construct: function(id, socket, path) {
     this._id = id;
     this._socket = socket;
+    this._path = path;
+    this._subscriptions = {};
   },
 
   members: {
     _id: null,
     _socket: null,
-    _subscriptions: {},
+    _path: null,
+    _subscriptions: null,
 
     getID: function() {
       return this._id;
@@ -47,7 +50,7 @@ JX.install('AphlictListener', {
     },
 
     getDescription: function() {
-      return 'Listener/' + this.getID();
+      return 'Listener/' + this.getID() + this._path;
     },
 
     writeMessage: function(message) {

support/aphlict/server/lib/AphlictListenerList.js

@@ -5,17 +5,22 @@ var JX = require('./javelin').JX;
 require('./AphlictListener');
 
 JX.install('AphlictListenerList', {
-  construct: function() {
+  construct: function(path) {
+    this._path = path;
     this._listeners = {};
   },
 
   members: {
     _listeners: null,
+    _path: null,
     _nextID: 0,
     _totalListenerCount: 0,
 
     addListener: function(socket) {
-      var listener = new JX.AphlictListener(this._generateNextID(), socket);
+      var listener = new JX.AphlictListener(
+        this._generateNextID(),
+        socket,
+        this._path);
 
       this._listeners[listener.getID()] = listener;
       this._totalListenerCount++;

webroot/rsrc/externals/javelin/lib/WebSocket.js

@@ -11,6 +11,7 @@ JX.install('WebSocket', {
 
   construct: function(uri) {
     this.setURI(uri);
+    this._resetDelay();
   },
 
   properties: {
@@ -70,7 +71,6 @@ JX.install('WebSocket', {
       }
 
       this._shouldClose = false;
-      this._resetDelay();
 
       this._socket = new WebSocket(this.getURI());
       this._socket.onopen = JX.bind(this, this._onopen);
@@ -171,6 +171,9 @@ JX.install('WebSocket', {
       // connection, the close handler will send us back here. We'll reconnect
       // more and more slowly until we eventually get a valid connection.
       this._delayUntilReconnect = this._delayUntilReconnect * 2;
+
+      // Max out at 5 minutes between attempts.
+      this._delayUntilReconnect = Math.min(this._delayUntilReconnect, 300000);
       this.open();
     }