Initial refactoring of sendmessage

js/modules/loki_message_api.js

@@ -31,24 +31,30 @@ const filterIncomingMessages = async messages => {
   return newMessages;
 };
 
-class LokiMessageAPI {
+const calcNonce = async (messageEventData, pubKey, data64, timestamp, ttl) => {
-  constructor({ snodeServerPort }) {
+  // Nonce is returned as a base64 string to include in header
-    this.snodeServerPort = snodeServerPort ? `:${snodeServerPort}` : '';
+  try {
-    this.jobQueue = new window.JobQueue();
+    window.Whisper.events.trigger('calculatingPoW', messageEventData);
-  }
+    const development = window.getEnvironment() !== 'production';
-
+    return callWorker(
-  async sendMessage(pubKey, data, messageTimeStamp, ttl, isPing = false) {
+      'calcPoW',
-    const timestamp = Date.now();
+      timestamp,
-
+      ttl,
-    // Data required to identify a message in a conversation
-    const messageEventData = {
       pubKey,
-      timestamp: messageTimeStamp,
+      data64,
-    };
+      development
+    );
+  } catch (err) {
+    // Something went horribly wrong
+    throw err;
+  }
+}
 
-    const data64 = dcodeIO.ByteBuffer.wrap(data).toString('base64');
+const trySendP2p = async (pubKey, data64, isPing, messageEventData) => {
   const p2pDetails = lokiP2pAPI.getContactP2pDetails(pubKey);
-    if (p2pDetails && (isPing || p2pDetails.isOnline)) {
+  if (!p2pDetails || (!isPing && !p2pDetails.isOnline)) {
+    return false;
+  }
   try {
     const port = p2pDetails.port ? `:${p2pDetails.port}` : '';
 
@@ -62,49 +68,49 @@ class LokiMessageAPI {
     } else {
       log.info(`Successful p2p message to ${pubKey}`);
     }
-        return;
+    return true;
   } catch (e) {
     lokiP2pAPI.setContactOffline(pubKey);
     if (isPing) {
       // If this was just a ping, we don't bother sending to storage server
       log.warn('Ping failed, contact marked offline', e);
-          return;
+      return true;
     }
     log.warn('Failed to send P2P message, falling back to storage', e);
+    return false;
   }
 }
 
-    // Nonce is returned as a base64 string to include in header
+class LokiMessageAPI {
-    let nonce;
+  constructor({ snodeServerPort }) {
-    try {
+    this.snodeServerPort = snodeServerPort ? `:${snodeServerPort}` : '';
-      window.Whisper.events.trigger('calculatingPoW', messageEventData);
+    this.jobQueue = new window.JobQueue();
-      const development = window.getEnvironment() !== 'production';
+    this.sendingSwarmNodes = {};
-      nonce = await callWorker(
-        'calcPoW',
-        timestamp,
-        ttl,
-        pubKey,
-        data64,
-        development
-      );
-    } catch (err) {
-      // Something went horribly wrong
-      throw err;
   }
 
-    const completedNodes = [];
+  async sendMessage(numConnections, pubKey, data, messageTimeStamp, ttl, isPing = false) {
-    const failedNodes = [];
+    // Data required to identify a message in a conversation
-    let successfulRequests = 0;
+    const messageEventData = {
-    let canResolve = true;
+      pubKey,
+      timestamp: messageTimeStamp,
+    };
 
-    let swarmNodes = await lokiSnodeAPI.getSwarmNodesForPubKey(pubKey);
+    const data64 = dcodeIO.ByteBuffer.wrap(data).toString('base64');
+    const p2pSuccess = await trySendP2p(pubKey, data64, isPing, messageEventData);
+    if (p2pSuccess) {
+      return;
+    }
 
-    const nodeComplete = nodeUrl => {
+    const timestamp = Date.now();
-      completedNodes.push(nodeUrl);
+    const nonce = await calcNonce(messageEventData, pubKey, data64, timestamp, ttl);
-      swarmNodes = swarmNodes.filter(node => node !== nodeUrl);
+    // Using timestamp as a unique identifier
-    };
+    this.sendingSwarmNodes[timestamp] = lokiSnodeAPI.getSwarmNodesForPubKey(pubKey);
+    if (this.sendingSwarmNodes[timestamp].length < numConnections) {
+      const freshNodes = await lokiSnodeAPI.getFreshSwarmNodes(pubKey);
+      await lokiSnodeAPI.updateSwarmNodes(pubKey, freshNodes);
+      this.sendingSwarmNodes[timestamp] = freshNodes;
+    }
 
-    const doRequest = async nodeUrl => {
     const params = {
       pubKey,
       ttl: ttl.toString(),
@@ -112,69 +118,68 @@ class LokiMessageAPI {
       timestamp: timestamp.toString(),
       data: data64,
     };
+    const promises = [];
+    for (let i = 0; i < numConnections; i += 1) {
+      promises.push(this.openSendConnection(params));
+    }
 
-      try {
+    const results = await Promise.all(promises);
-        await rpc(`http://${nodeUrl}`, this.snodeServerPort, 'store', params);
+    delete this.sendingSwarmNodes[timestamp];
+    if (results.every(value => value === false)) {
+      throw new window.textsecure.EmptySwarmError(
+        pubKey,
+        'Ran out of swarm nodes to query'
+      );
+    }
+    if (results.every(value => value === true)) {
+      log.info(`Successful storage message to ${pubKey}`);
+    } else {
+      log.warn(`Partially successful storage message to ${pubKey}`);
+    }
+  }
 
-        nodeComplete(nodeUrl);
+  async openSendConnection(params) {
-        successfulRequests += 1;
+    while (!_.isEmpty(this.sendingSwarmNodes[params.timestamp])) {
+      const url = this.sendingSwarmNodes[params.timestamp].shift();
+      const successfulSend = await this.sendToNode(url, params);
+      if (successfulSend) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  async sendToNode(url, params) {
+    let successiveFailures = 0;
+    while (successiveFailures < 3) {
+      await sleepFor(successiveFailures * 500);
+      try {
+        await rpc(`http://${url}`, this.snodeServerPort, 'store', params);
+        return true;
       } catch (e) {
         log.warn('Loki send message:', e);
         if (e instanceof textsecure.WrongSwarmError) {
           const { newSwarm } = e;
-          await lokiSnodeAPI.updateSwarmNodes(pubKey, newSwarm);
+          await lokiSnodeAPI.updateSwarmNodes(params.pubKey, newSwarm);
-          completedNodes.push(nodeUrl);
+          this.sendingSwarmNodes[params.timestamp] = newSwarm;
+          return false;
         } else if (e instanceof textsecure.NotFoundError) {
-          canResolve = false;
+          // TODO: Handle resolution error
+          successiveFailures += 1;
         } else if (e instanceof textsecure.HTTPError) {
-          // We mark the node as complete as we could still reach it
+          // TODO: Handle working connection but error response
-          nodeComplete(nodeUrl);
+          successiveFailures += 1;
         } else {
-          const removeNode = await lokiSnodeAPI.unreachableNode(
+          successiveFailures += 1;
-            pubKey,
-            nodeUrl
-          );
-          if (removeNode) {
-            log.error('Loki send message:', e);
-            nodeComplete(nodeUrl);
-            failedNodes.push(nodeUrl);
-          }
-        }
-      }
-    };
-
-    while (successfulRequests < MINIMUM_SUCCESSFUL_REQUESTS) {
-      if (!canResolve) {
-        throw new window.textsecure.DNSResolutionError('Sending messages');
-      }
-      if (swarmNodes.length === 0) {
-        const freshNodes = await lokiSnodeAPI.getFreshSwarmNodes(pubKey);
-        const goodNodes = _.difference(freshNodes, failedNodes);
-        await lokiSnodeAPI.updateSwarmNodes(pubKey, goodNodes);
-        swarmNodes = _.difference(freshNodes, completedNodes);
-        if (swarmNodes.length === 0) {
-          if (successfulRequests !== 0) {
-            // TODO: Decide how to handle some completed requests but not enough
-            log.warn(`Partially successful storage message to ${pubKey}`);
-            return;
         }
-          throw new window.textsecure.EmptySwarmError(
-            pubKey,
-            'Ran out of swarm nodes to query'
-          );
       }
     }
-
+    log.error(`Failed to send to node: ${url}`);
-      const remainingRequests =
+    await lokiSnodeAPI.unreachableNode(
-        MINIMUM_SUCCESSFUL_REQUESTS - successfulRequests;
+      params.pubKey,
-
+      url
-      await Promise.all(
-        swarmNodes
-          .splice(0, remainingRequests)
-          .map(nodeUrl => doRequest(nodeUrl))
     );
-    }
+    return false;
-    log.info(`Successful storage message to ${pubKey}`);
   }
 
   async retrieveNextMessages(nodeUrl, nodeData, ourKey) {

js/modules/loki_snode_api.js

@@ -73,29 +73,10 @@ class LokiSnodeAPI {
 
   async unreachableNode(pubKey, nodeUrl) {
     if (pubKey === window.textsecure.storage.user.getNumber()) {
-      if (!this.ourSwarmNodes[nodeUrl]) {
-        this.ourSwarmNodes[nodeUrl] = {
-          failureCount: 1,
-        };
-      } else {
-        this.ourSwarmNodes[nodeUrl].failureCount += 1;
-      }
-      if (this.ourSwarmNodes[nodeUrl].failureCount < FAILURE_THRESHOLD) {
-        return false;
-      }
       delete this.ourSwarmNodes[nodeUrl];
-      return true;
+      return;
-    }
-    if (!this.contactSwarmNodes[nodeUrl]) {
-      this.contactSwarmNodes[nodeUrl] = {
-        failureCount: 1,
-      };
-    } else {
-      this.contactSwarmNodes[nodeUrl].failureCount += 1;
-    }
-    if (this.contactSwarmNodes[nodeUrl].failureCount < FAILURE_THRESHOLD) {
-      return false;
     }
+
     const conversation = ConversationController.get(pubKey);
     const swarmNodes = [...conversation.get('swarmNodes')];
     if (swarmNodes.includes(nodeUrl)) {
@@ -103,14 +84,12 @@ class LokiSnodeAPI {
       await conversation.updateSwarmNodes(filteredNodes);
       delete this.contactSwarmNodes[nodeUrl];
     }
-    return true;
   }
 
   async updateLastHash(nodeUrl, lastHash, expiresAt) {
     await window.Signal.Data.updateLastHash({ nodeUrl, lastHash, expiresAt });
     if (!this.ourSwarmNodes[nodeUrl]) {
       this.ourSwarmNodes[nodeUrl] = {
-        failureCount: 0,
         lastHash,
       };
     } else {
@@ -118,7 +97,7 @@ class LokiSnodeAPI {
     }
   }
 
-  async getSwarmNodesForPubKey(pubKey) {
+  getSwarmNodesForPubKey(pubKey) {
     try {
       const conversation = ConversationController.get(pubKey);
       const swarmNodes = [...conversation.get('swarmNodes')];
@@ -146,7 +125,6 @@ class LokiSnodeAPI {
     const ps = newNodes.map(async url => {
       const lastHash = await window.Signal.Data.getLastHashBySnode(url);
       this.ourSwarmNodes[url] = {
-        failureCount: 0,
         lastHash,
       };
     });

libtextsecure/outgoing_message.js

@@ -187,7 +187,9 @@ OutgoingMessage.prototype = {
   async transmitMessage(number, data, timestamp, ttl = 24 * 60 * 60 * 1000) {
     const pubKey = number;
     try {
+      // TODO: Make NUM_CONCURRENT_CONNECTIONS a global constant
       await lokiMessageAPI.sendMessage(
+        2,
         pubKey,
         data,
         timestamp,