Retrieve profiles in parallel.

app/src/main/java/org/thoughtcrime/securesms/contacts/sync/DirectoryHelperV1.java

@@ -56,6 +56,9 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 class DirectoryHelperV1 {
 
@@ -339,10 +342,16 @@ class DirectoryHelperV1 {
 
   private static boolean isUuidRegistered(@NonNull Context context, @NonNull Recipient recipient) throws IOException {
     try {
-      ProfileUtil.retrieveProfile(context, recipient, SignalServiceProfile.RequestType.PROFILE);
+      ProfileUtil.retrieveProfile(context, recipient, SignalServiceProfile.RequestType.PROFILE).get(10, TimeUnit.SECONDS);
       return true;
-    } catch (NotFoundException e) {
+    } catch (ExecutionException e) {
-      return false;
+      if (e.getCause() instanceof NotFoundException) {
+        return false;
+      } else {
+        throw new IOException(e);
+      }
+    } catch (InterruptedException | TimeoutException e) {
+      throw new IOException(e);
     }
   }
 

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationActivity.java

@@ -207,7 +207,6 @@ import org.thoughtcrime.securesms.recipients.RecipientExporter;
 import org.thoughtcrime.securesms.recipients.RecipientFormattingException;
 import org.thoughtcrime.securesms.recipients.RecipientId;
 import org.thoughtcrime.securesms.recipients.RecipientUtil;
-import org.thoughtcrime.securesms.recipients.ui.bottomsheet.RecipientBottomSheetDialogFragment;
 import org.thoughtcrime.securesms.registration.RegistrationNavigationActivity;
 import org.thoughtcrime.securesms.service.KeyCachingService;
 import org.thoughtcrime.securesms.sms.MessageSender;
@@ -1957,7 +1956,7 @@ public class ConversationActivity extends PassphraseRequiredActionBarActivity
       return;
     }
 
-    ApplicationDependencies.getJobManager().add(RetrieveProfileJob.forRecipient(recipient.get()));
+    RetrieveProfileJob.enqueueAsync(recipient.getId());
   }
 
   private void onRecipientChanged(@NonNull Recipient recipient) {

app/src/main/java/org/thoughtcrime/securesms/groups/GroupsV2CapabilityChecker.java

@@ -43,7 +43,7 @@ final class GroupsV2CapabilityChecker {
       Recipient.Capability gv2Capability  = member.getGroupsV2Capability();
 
       if (gv2Capability != Recipient.Capability.SUPPORTED) {
-        if (!ApplicationDependencies.getJobManager().runSynchronously(RetrieveProfileJob.forRecipient(member), TimeUnit.SECONDS.toMillis(1000)).isPresent()) {
+        if (!ApplicationDependencies.getJobManager().runSynchronously(RetrieveProfileJob.forRecipient(member.getId()), TimeUnit.SECONDS.toMillis(1000)).isPresent()) {
           throw new IOException("Recipient capability was not retrieved in time");
         }
       }

app/src/main/java/org/thoughtcrime/securesms/groups/v2/GroupCandidateHelper.java

@@ -59,17 +59,12 @@ public final class GroupCandidateHelper {
       ProfileKey profileKey = ProfileKeyUtil.profileKeyOrNull(recipient.getProfileKey());
 
       if (profileKey != null) {
-        try {
+        Optional<ProfileKeyCredential> profileKeyCredentialOptional = signalServiceAccountManager.resolveProfileKeyCredential(uuid, profileKey);
-          Optional<ProfileKeyCredential> profileKeyCredentialOptional = signalServiceAccountManager.resolveProfileKeyCredential(uuid, profileKey);
 
-          if (profileKeyCredentialOptional.isPresent()) {
+        if (profileKeyCredentialOptional.isPresent()) {
-            candidate = candidate.withProfileKeyCredential(profileKeyCredentialOptional.get());
+          candidate = candidate.withProfileKeyCredential(profileKeyCredentialOptional.get());
 
-            recipientDatabase.setProfileKeyCredential(recipient.getId(), profileKey, profileKeyCredentialOptional.get());
+          recipientDatabase.setProfileKeyCredential(recipient.getId(), profileKey, profileKeyCredentialOptional.get());
-          }
-        } catch (VerificationFailedException e) {
-          Log.w(TAG, e);
-          throw new IOException(e);
         }
       }
     }

app/src/main/java/org/thoughtcrime/securesms/groups/v2/processing/GroupsV2StateProcessor.java

@@ -290,9 +290,7 @@ public final class GroupsV2StateProcessor {
 
       if (!updated.isEmpty()) {
         Log.i(TAG, String.format(Locale.US, "Learned %d new profile keys, scheduling profile retrievals", updated.size()));
-        for (RecipientId recipient : updated) {
+        RetrieveProfileJob.enqueue(updated);
-          ApplicationDependencies.getJobManager().add(RetrieveProfileJob.forRecipient(recipient));
-        }
       }
     }
 

app/src/main/java/org/thoughtcrime/securesms/jobmanager/Data.java

@@ -3,9 +3,13 @@ package org.thoughtcrime.securesms.jobmanager;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 
+import com.annimon.stream.Stream;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 public class Data {
@@ -76,6 +80,14 @@ public class Data {
     return stringArrays.get(key);
   }
 
+  /**
+   * Helper method for {@link #getStringArray(String)} that returns the value as a list.
+   */
+  public List<String> getStringArrayAsList(@NonNull String key) {
+    throwIfAbsent(stringArrays, key);
+    return Arrays.asList(stringArrays.get(key));
+  }
+
 
   public boolean hasInt(@NonNull String key) {
     return integers.containsKey(key);
@@ -255,6 +267,14 @@ public class Data {
       return this;
     }
 
+    /**
+     * Helper method for {@link #putStringArray(String, String[])} that takes a list.
+     */
+    public Builder putStringListAsArray(@NonNull String key, @NonNull List<String> value) {
+      stringArrays.put(key, value.toArray(new String[0]));
+      return this;
+    }
+
     public Builder putInt(@NonNull String key, int value) {
       integers.put(key, value);
       return this;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobManager.java

@@ -39,7 +39,7 @@ public class JobManager implements ConstraintObserver.Notifier {
 
   private static final String TAG = JobManager.class.getSimpleName();
 
-  public static final int CURRENT_VERSION = 6;
+  public static final int CURRENT_VERSION = 7;
 
   private final Application     application;
   private final Configuration   configuration;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/migrations/RetrieveProfileJobMigration.java

@@ -0,0 +1,46 @@
+package org.thoughtcrime.securesms.jobmanager.migrations;
+
+import androidx.annotation.NonNull;
+
+import org.thoughtcrime.securesms.database.MmsSmsDatabase;
+import org.thoughtcrime.securesms.jobmanager.Data;
+import org.thoughtcrime.securesms.jobmanager.JobMigration;
+import org.thoughtcrime.securesms.logging.Log;
+
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class RetrieveProfileJobMigration extends JobMigration {
+
+  private static final String TAG = Log.tag(RetrieveProfileJobMigration.class);
+
+  public RetrieveProfileJobMigration() {
+    super(7);
+  }
+
+  @Override
+  protected @NonNull JobData migrate(@NonNull JobData jobData) {
+    Log.i(TAG, "Running.");
+
+    if ("RetrieveProfileJob".equals(jobData.getFactoryKey())) {
+      return migrateRetrieveProfileJob(jobData);
+    }
+    return jobData;
+  }
+
+  private static @NonNull JobData migrateRetrieveProfileJob(@NonNull JobData jobData) {
+    Data data = jobData.getData();
+
+    if (data.hasString("recipient")) {
+      Log.i(TAG, "Migrating job.");
+
+      String recipient = data.getString("recipient");
+      return jobData.withData(new Data.Builder()
+                                      .putStringArray("recipients", new String[] { recipient })
+                                      .build());
+    } else {
+      return jobData;
+    }
+  }
+
+}

app/src/main/java/org/thoughtcrime/securesms/jobs/JobManagerFactories.java

@@ -21,6 +21,7 @@ import org.thoughtcrime.securesms.jobmanager.migrations.PushProcessMessageQueueJ
 import org.thoughtcrime.securesms.jobmanager.migrations.RecipientIdFollowUpJobMigration;
 import org.thoughtcrime.securesms.jobmanager.migrations.RecipientIdFollowUpJobMigration2;
 import org.thoughtcrime.securesms.jobmanager.migrations.RecipientIdJobMigration;
+import org.thoughtcrime.securesms.jobmanager.migrations.RetrieveProfileJobMigration;
 import org.thoughtcrime.securesms.jobmanager.migrations.SendReadReceiptsJobMigration;
 import org.thoughtcrime.securesms.migrations.AvatarIdRemovalMigrationJob;
 import org.thoughtcrime.securesms.migrations.AvatarMigrationJob;
@@ -169,6 +170,7 @@ public final class JobManagerFactories {
                          new RecipientIdFollowUpJobMigration(),
                          new RecipientIdFollowUpJobMigration2(),
                          new SendReadReceiptsJobMigration(DatabaseFactory.getMmsSmsDatabase(application)),
-                         new PushProcessMessageQueueJobMigration(application));
+                         new PushProcessMessageQueueJobMigration(application),
+                         new RetrieveProfileJobMigration());
   }
 }

app/src/main/java/org/thoughtcrime/securesms/jobs/PushProcessMessageJob.java

@@ -69,7 +69,6 @@ import org.thoughtcrime.securesms.mms.OutgoingSecureMediaMessage;
 import org.thoughtcrime.securesms.mms.QuoteModel;
 import org.thoughtcrime.securesms.mms.SlideDeck;
 import org.thoughtcrime.securesms.mms.StickerSlide;
-import org.thoughtcrime.securesms.notifications.DefaultMessageNotifier;
 import org.thoughtcrime.securesms.notifications.MessageNotifier;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientId;
@@ -1450,7 +1449,7 @@ public final class PushProcessMessageJob extends BaseJob {
 
     if (messageProfileKey != null) {
       if (database.setProfileKey(recipient.getId(), messageProfileKey)) {
-        ApplicationDependencies.getJobManager().add(RetrieveProfileJob.forRecipient(recipient));
+        ApplicationDependencies.getJobManager().add(RetrieveProfileJob.forRecipient(recipient.getId()));
       }
     } else {
       Log.w(TAG, "Ignored invalid profile key seen in message");

app/src/main/java/org/thoughtcrime/securesms/jobs/RefreshOwnProfileJob.java

@@ -72,7 +72,7 @@ public class RefreshOwnProfileJob extends BaseJob {
     }
 
     Recipient            self                 = Recipient.self();
-    ProfileAndCredential profileAndCredential = ProfileUtil.retrieveProfile(context, self, getRequestType(self));
+    ProfileAndCredential profileAndCredential = ProfileUtil.retrieveProfileSync(context, self, getRequestType(self));
     SignalServiceProfile profile              = profileAndCredential.getProfile();
 
     setProfileName(profile.getName());

app/src/main/java/org/thoughtcrime/securesms/jobs/RetrieveProfileJob.java

@@ -1,10 +1,14 @@
 package org.thoughtcrime.securesms.jobs;
 
 
+import android.content.Context;
 import android.text.TextUtils;
 
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
+import androidx.annotation.WorkerThread;
+
+import com.annimon.stream.Stream;
 
 import org.signal.zkgroup.profiles.ProfileKey;
 import org.signal.zkgroup.profiles.ProfileKeyCredential;
@@ -16,31 +20,49 @@ import org.thoughtcrime.securesms.database.RecipientDatabase.UnidentifiedAccessM
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.jobmanager.Data;
 import org.thoughtcrime.securesms.jobmanager.Job;
+import org.thoughtcrime.securesms.jobmanager.JobManager;
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
+import org.thoughtcrime.securesms.linkpreview.Link;
 import org.thoughtcrime.securesms.logging.Log;
 import org.thoughtcrime.securesms.profiles.ProfileName;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientId;
+import org.thoughtcrime.securesms.transport.RetryLaterException;
 import org.thoughtcrime.securesms.util.Base64;
 import org.thoughtcrime.securesms.util.FeatureFlags;
 import org.thoughtcrime.securesms.util.IdentityUtil;
 import org.thoughtcrime.securesms.util.ProfileUtil;
+import org.thoughtcrime.securesms.util.Stopwatch;
 import org.thoughtcrime.securesms.util.Util;
+import org.thoughtcrime.securesms.util.concurrent.SignalExecutors;
 import org.whispersystems.libsignal.IdentityKey;
 import org.whispersystems.libsignal.InvalidKeyException;
+import org.whispersystems.libsignal.util.Pair;
 import org.whispersystems.libsignal.util.guava.Optional;
 import org.whispersystems.signalservice.api.crypto.InvalidCiphertextException;
 import org.whispersystems.signalservice.api.crypto.ProfileCipher;
 import org.whispersystems.signalservice.api.profiles.ProfileAndCredential;
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfile;
+import org.whispersystems.signalservice.api.push.exceptions.NotFoundException;
+import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException;
+import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 /**
  * Retrieves a users profile and sets the appropriate local fields.
- * <p>
- * Recipient can be self if you use {@link #forRecipient} and it will delegate to {@link RefreshOwnProfileJob}.
  */
 public class RetrieveProfileJob extends BaseJob {
 
@@ -48,38 +70,84 @@ public class RetrieveProfileJob extends BaseJob {
 
   private static final String TAG = RetrieveProfileJob.class.getSimpleName();
 
-  private static final String KEY_RECIPIENT = "recipient";
+  private static final String KEY_RECIPIENTS = "recipients";
 
-  private final RecipientId recipientId;
+  private final List<RecipientId> recipientIds;
 
-  public static Job forRecipient(@NonNull Recipient recipient) {
+  /**
-    return forRecipient(recipient.getId());
+   * Identical to {@link #enqueue(Collection)})}, but run on a background thread for convenience.
+   */
+  public static void enqueueAsync(@NonNull RecipientId recipientId) {
+    SignalExecutors.BOUNDED.execute(() -> {
+      ApplicationDependencies.getJobManager().add(forRecipient(recipientId));
+    });
   }
 
-  public static Job forRecipient(@NonNull RecipientId recipientId) {
+  /**
-    if (Recipient.self().getId().equals(recipientId)) {
+   * Submits the necessary jobs to refresh the profiles of the requested recipients. Works for any
+   * RecipientIds, including individuals, groups, or yourself.
+   */
+  @WorkerThread
+  public static void enqueue(@NonNull Collection<RecipientId> recipientIds) {
+    Context           context    = ApplicationDependencies.getApplication();
+    JobManager        jobManager = ApplicationDependencies.getJobManager();
+    List<RecipientId> combined   = new LinkedList<>();
+
+    for (RecipientId recipientId : recipientIds) {
+      Recipient recipient = Recipient.resolved(recipientId);
+
+      if (recipient.isLocalNumber()) {
+        jobManager.add(new RefreshOwnProfileJob());
+      } else if (recipient.isGroup()) {
+        List<Recipient> recipients = DatabaseFactory.getGroupDatabase(context).getGroupMembers(recipient.requireGroupId(), GroupDatabase.MemberSet.FULL_MEMBERS_EXCLUDING_SELF);
+        combined.addAll(Stream.of(recipients).map(Recipient::getId).toList());
+      } else {
+        combined.add(recipientId);
+      }
+    }
+
+    jobManager.add(new RetrieveProfileJob(combined));
+  }
+
+  /**
+   * Works for any RecipientId, whether it's an individual, group, or yourself.
+   */
+  @WorkerThread
+  public static @NonNull Job forRecipient(@NonNull RecipientId recipientId) {
+    Recipient recipient = Recipient.resolved(recipientId);
+
+    if (recipient.isLocalNumber()) {
       return new RefreshOwnProfileJob();
+    } else if (recipient.isGroup()) {
+      Context         context    = ApplicationDependencies.getApplication();
+      List<Recipient> recipients = DatabaseFactory.getGroupDatabase(context).getGroupMembers(recipient.requireGroupId(), GroupDatabase.MemberSet.FULL_MEMBERS_EXCLUDING_SELF);
+
+      return new RetrieveProfileJob(Stream.of(recipients).map(Recipient::getId).toList());
     } else {
-      return new RetrieveProfileJob(recipientId);
+      return new RetrieveProfileJob(Collections.singletonList(recipientId));
     }
   }
 
-  private RetrieveProfileJob(@NonNull RecipientId recipientId) {
+  private RetrieveProfileJob(@NonNull List<RecipientId> recipientIds) {
     this(new Job.Parameters.Builder()
                            .addConstraint(NetworkConstraint.KEY)
                            .setMaxAttempts(3)
                            .build(),
-         recipientId);
+         recipientIds);
   }
 
-  private RetrieveProfileJob(@NonNull Job.Parameters parameters, @NonNull RecipientId recipientId) {
+  private RetrieveProfileJob(@NonNull Job.Parameters parameters, @NonNull List<RecipientId> recipientIds) {
     super(parameters);
-    this.recipientId = recipientId;
+    this.recipientIds = recipientIds;
   }
 
   @Override
   public @NonNull Data serialize() {
-    return new Data.Builder().putString(KEY_RECIPIENT, recipientId.serialize()).build();
+    return new Data.Builder()
+                   .putStringListAsArray(KEY_RECIPIENTS, Stream.of(recipientIds)
+                                                               .map(RecipientId::serialize)
+                                                               .toList())
+                   .build();
   }
 
   @Override
@@ -88,39 +156,74 @@ public class RetrieveProfileJob extends BaseJob {
   }
 
   @Override
-  public void onRun() throws IOException {
+  public void onRun() throws IOException, RetryLaterException {
-    Log.i(TAG, "Retrieving profile of " + recipientId);
+    Stopwatch        stopwatch = new Stopwatch("RetrieveProfile");
+    Set<RecipientId> retries   = new HashSet<>();
 
-    Recipient resolved = Recipient.resolved(recipientId);
+    List<Recipient> recipients = Stream.of(recipientIds).map(Recipient::resolved).toList();
+    stopwatch.split("resolve");
 
-    if (resolved.isGroup()) handleGroupRecipient(resolved);
+    List<Pair<Recipient, ListenableFuture<ProfileAndCredential>>> futures = Stream.of(recipients)
-    else                    handleIndividualRecipient(resolved);
+                                                                                  .filter(Recipient::hasServiceIdentifier)
+                                                                                  .map(r -> new Pair<>(r, ProfileUtil.retrieveProfile(context, r, getRequestType(r))))
+                                                                                  .toList();
+    stopwatch.split("futures");
+
+    List<Pair<Recipient, ProfileAndCredential>> profiles = Stream.of(futures)
+                                                                 .map(pair -> {
+                                                                   Recipient recipient = pair.first();
+
+                                                                   try {
+                                                                     ProfileAndCredential profile = pair.second().get(5, TimeUnit.SECONDS);
+                                                                     return new Pair<>(recipient, profile);
+                                                                   } catch (InterruptedException | TimeoutException e) {
+                                                                     retries.add(recipient.getId());
+                                                                   } catch (ExecutionException e) {
+                                                                     if (e.getCause() instanceof PushNetworkException) {
+                                                                       retries.add(recipient.getId());
+                                                                     } else if (e.getCause() instanceof NotFoundException) {
+                                                                       Log.w(TAG, "Failed to find a profile for " + recipient.getId());
+                                                                     } else {
+                                                                       Log.w(TAG, "Failed to retrieve profile for " + recipient.getId());
+                                                                     }
+                                                                   }
+                                                                   return null;
+                                                                 })
+                                                                 .withoutNulls()
+                                                                 .toList();
+    stopwatch.split("network");
+
+    for (Pair<Recipient, ProfileAndCredential> profile : profiles) {
+      process(profile.first(), profile.second());
+    }
+
+    stopwatch.split("process");
+
+    long keyCount = Stream.of(profiles).map(Pair::first).map(Recipient::getProfileKey).withoutNulls().count();
+    Log.d(TAG, String.format(Locale.US, "Started with %d recipient(s). Found %d profile(s), and had keys for %d of them. Will retry %d.", recipients.size(), profiles.size(), keyCount, retries.size()));
+
+    stopwatch.stop(TAG);
+
+    recipientIds.clear();
+    recipientIds.addAll(retries);
+
+    if (recipientIds.size() > 0) {
+      throw new RetryLaterException();
+    }
   }
 
   @Override
   public boolean onShouldRetry(@NonNull Exception e) {
-    return false;
+    return e instanceof RetryLaterException;
   }
 
   @Override
   public void onFailure() {}
 
-  private void handleIndividualRecipient(Recipient recipient) throws IOException {
+  private void process(Recipient recipient, ProfileAndCredential profileAndCredential) throws IOException {
-     if (recipient.hasServiceIdentifier()) handlePhoneNumberRecipient(recipient);
-     else                                  Log.w(TAG, "Skipping fetching profile of non-Signal recipient");
-  }
-
-  private void handlePhoneNumberRecipient(Recipient recipient) throws IOException {
-    ProfileAndCredential profileAndCredential = ProfileUtil.retrieveProfile(context, recipient, getRequestType(recipient));
     SignalServiceProfile profile              = profileAndCredential.getProfile();
     ProfileKey           recipientProfileKey  = ProfileKeyUtil.profileKeyOrNull(recipient.getProfileKey());
 
-    if (recipientProfileKey == null) {
-      Log.i(TAG, "No profile key available for " + recipient.getId());
-    } else {
-      Log.i(TAG, "Profile key available for " + recipient.getId());
-    }
-
     setProfileName(recipient, profile.getName());
     setProfileAvatar(recipient, profile.getAvatar());
     if (FeatureFlags.usernames()) setUsername(recipient, profile.getUsername());
@@ -150,14 +253,6 @@ public class RetrieveProfileJob extends BaseJob {
            : SignalServiceProfile.RequestType.PROFILE;
   }
 
-  private void handleGroupRecipient(Recipient group) throws IOException {
-    List<Recipient> recipients = DatabaseFactory.getGroupDatabase(context).getGroupMembers(group.requireGroupId(), GroupDatabase.MemberSet.FULL_MEMBERS_EXCLUDING_SELF);
-
-    for (Recipient recipient : recipients) {
-      handleIndividualRecipient(recipient);
-    }
-  }
-
   private void setIdentityKey(Recipient recipient, String identityKeyValue) {
     try {
       if (TextUtils.isEmpty(identityKeyValue)) {
@@ -222,7 +317,7 @@ public class RetrieveProfileJob extends BaseJob {
 
       String plaintextProfileName = ProfileUtil.decryptName(profileKey, profileName);
 
-      if (!Util.equals(plaintextProfileName, recipient.getProfileName().serialize())) {
+      if (!Objects.equals(plaintextProfileName, recipient.getProfileName().serialize())) {
         Log.i(TAG, "Profile name updated. Writing new value.");
         DatabaseFactory.getRecipientDatabase(context).setProfileName(recipient.getId(), ProfileName.fromSerialized(plaintextProfileName));
       }
@@ -230,7 +325,9 @@ public class RetrieveProfileJob extends BaseJob {
       if (TextUtils.isEmpty(plaintextProfileName)) {
         Log.i(TAG, "No profile name set.");
       }
-    } catch (InvalidCiphertextException | IOException e) {
+    } catch (InvalidCiphertextException e) {
+      Log.w(TAG, "Bad profile key for " + recipient.getId());
+    } catch (IOException e) {
       Log.w(TAG, e);
     }
   }
@@ -240,8 +337,6 @@ public class RetrieveProfileJob extends BaseJob {
 
     if (!Util.equals(profileAvatar, recipient.getProfileAvatar())) {
       ApplicationDependencies.getJobManager().add(new RetrieveProfileAvatarJob(recipient, profileAvatar));
-    } else {
-      Log.d(TAG, "Skipping avatar fetch for " + recipient.getId());
     }
   }
 
@@ -261,7 +356,10 @@ public class RetrieveProfileJob extends BaseJob {
 
     @Override
     public @NonNull RetrieveProfileJob create(@NonNull Parameters parameters, @NonNull Data data) {
-      return new RetrieveProfileJob(parameters, RecipientId.from(data.getString(KEY_RECIPIENT)));
+      String[]          ids          = data.getStringArray(KEY_RECIPIENTS);
+      List<RecipientId> recipientIds = Stream.of(ids).map(RecipientId::from).toList();
+
+      return new RetrieveProfileJob(parameters, recipientIds);
     }
   }
 }

app/src/main/java/org/thoughtcrime/securesms/profiles/ProfileName.java

@@ -56,9 +56,11 @@ public final class ProfileName implements Parcelable {
   public @NonNull String serialize() {
     if (isGivenNameEmpty()) {
       return "";
+    } else if (familyName.isEmpty()) {
+      return givenName;
+    } else {
+      return String.format("%s\0%s", givenName, familyName);
     }
-
-    return String.format("%s\0%s", givenName, familyName);
   }
 
   @Override

app/src/main/java/org/thoughtcrime/securesms/profiles/edit/EditSelfProfileRepository.java

@@ -33,6 +33,8 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 class EditSelfProfileRepository implements EditProfileRepository {
 
@@ -149,10 +151,10 @@ class EditSelfProfileRepository implements EditProfileRepository {
   @WorkerThread
   private @NonNull Optional<String> getUsernameInternal() {
     try {
-      SignalServiceProfile profile = ProfileUtil.retrieveProfile(context, Recipient.self(), SignalServiceProfile.RequestType.PROFILE).getProfile();
+      SignalServiceProfile profile = ProfileUtil.retrieveProfile(context, Recipient.self(), SignalServiceProfile.RequestType.PROFILE).get(5, TimeUnit.SECONDS).getProfile();
       TextSecurePreferences.setLocalUsername(context, profile.getUsername());
       DatabaseFactory.getRecipientDatabase(context).setUsername(Recipient.self().getId(), profile.getUsername());
-    } catch (IOException e) {
+    } catch (TimeoutException | InterruptedException | ExecutionException e) {
       Log.w(TAG, "Failed to retrieve username remotely! Using locally-cached version.");
     }
     return Optional.fromNullable(TextSecurePreferences.getLocalUsername(context));

app/src/main/java/org/thoughtcrime/securesms/util/IdentityUtil.java

@@ -123,16 +123,17 @@ public class IdentityUtil {
     long                 time          = System.currentTimeMillis();
     SmsDatabase          smsDatabase   = DatabaseFactory.getSmsDatabase(context);
     GroupDatabase        groupDatabase = DatabaseFactory.getGroupDatabase(context);
-    GroupDatabase.Reader reader        = groupDatabase.getGroups();
 
-    GroupDatabase.GroupRecord groupRecord;
+    try (GroupDatabase.Reader reader = groupDatabase.getGroups()) {
+      GroupDatabase.GroupRecord groupRecord;
 
-    while ((groupRecord = reader.getNext()) != null) {
+      while ((groupRecord = reader.getNext()) != null) {
-      if (groupRecord.getMembers().contains(recipientId) && groupRecord.isActive()) {
+        if (groupRecord.getMembers().contains(recipientId) && groupRecord.isActive()) {
-        IncomingTextMessage           incoming    = new IncomingTextMessage(recipientId, 1, time, time, null, Optional.of(groupRecord.getId()), 0, false);
+          IncomingTextMessage           incoming    = new IncomingTextMessage(recipientId, 1, time, time, null, Optional.of(groupRecord.getId()), 0, false);
-        IncomingIdentityUpdateMessage groupUpdate = new IncomingIdentityUpdateMessage(incoming);
+          IncomingIdentityUpdateMessage groupUpdate = new IncomingIdentityUpdateMessage(incoming);
 
-        smsDatabase.insertMessageInbox(groupUpdate);
+          smsDatabase.insertMessageInbox(groupUpdate);
+        }
       }
     }
 

app/src/main/java/org/thoughtcrime/securesms/util/ProfileUtil.java

@@ -26,8 +26,16 @@ import org.whispersystems.signalservice.api.profiles.ProfileAndCredential;
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfile;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
 import org.whispersystems.signalservice.api.push.exceptions.NonSuccessfulResponseCodeException;
+import org.whispersystems.signalservice.api.push.exceptions.NotFoundException;
+import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException;
+import org.whispersystems.signalservice.internal.util.concurrent.CascadingFuture;
+import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 /**
  * Aids in the retrieval and decryption of profiles.
@@ -37,31 +45,44 @@ public final class ProfileUtil {
   private ProfileUtil() {
   }
 
-  private static final String TAG = Log.tag(ProfileUtil.class);
-
   @WorkerThread
-  public static @NonNull ProfileAndCredential retrieveProfile(@NonNull Context context,
+  public static @NonNull ProfileAndCredential retrieveProfileSync(@NonNull Context context,
-                                                              @NonNull Recipient recipient,
+                                                                  @NonNull Recipient recipient,
-                                                              @NonNull SignalServiceProfile.RequestType requestType)
+                                                                  @NonNull SignalServiceProfile.RequestType requestType)
-    throws IOException
+      throws IOException
+  {
+    try {
+      return retrieveProfile(context, recipient, requestType).get(10, TimeUnit.SECONDS);
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof PushNetworkException) {
+        throw (PushNetworkException) e.getCause();
+      } else {
+        throw new IOException(e);
+      }
+    } catch (InterruptedException | TimeoutException e) {
+      throw new PushNetworkException(e);
+    }
+  }
+
+  public static @NonNull ListenableFuture<ProfileAndCredential> retrieveProfile(@NonNull Context context,
+                                                                                @NonNull Recipient recipient,
+                                                                                @NonNull SignalServiceProfile.RequestType requestType)
   {
     SignalServiceAddress         address            = RecipientUtil.toSignalServiceAddress(context, recipient);
     Optional<UnidentifiedAccess> unidentifiedAccess = getUnidentifiedAccess(context, recipient);
     Optional<ProfileKey>         profileKey         = ProfileKeyUtil.profileKeyOptional(recipient.getProfileKey());
 
-    ProfileAndCredential profile;
+    if (unidentifiedAccess.isPresent()) {
-
+      return new CascadingFuture<>(Arrays.asList(() -> getPipeRetrievalFuture(address, profileKey, unidentifiedAccess, requestType),
-    try {
+                                                 () -> getSocketRetrievalFuture(address, profileKey, unidentifiedAccess, requestType),
-      profile = retrieveProfileInternal(address, profileKey, unidentifiedAccess, requestType);
+                                                 () -> getPipeRetrievalFuture(address, profileKey, Optional.absent(), requestType),
-    } catch (NonSuccessfulResponseCodeException e) {
+                                                 () -> getSocketRetrievalFuture(address, profileKey, Optional.absent(), requestType)),
-      if (unidentifiedAccess.isPresent()) {
+                                   e -> !(e instanceof NotFoundException));
-        profile = retrieveProfileInternal(address, profileKey, Optional.absent(), requestType);
+    } else {
-      } else {
+      return new CascadingFuture<>(Arrays.asList(() -> getPipeRetrievalFuture(address, profileKey, Optional.absent(), requestType),
-        throw e;
+                                                 () -> getSocketRetrievalFuture(address, profileKey, Optional.absent(), requestType)),
-      }
+                                   e -> !(e instanceof NotFoundException));
     }
-
-    return profile;
   }
 
   public static @Nullable String decryptName(@NonNull ProfileKey profileKey, @Nullable String encryptedName)
@@ -75,32 +96,30 @@ public final class ProfileUtil {
     return new String(profileCipher.decryptName(Base64.decode(encryptedName)));
   }
 
-  @WorkerThread
+  private static @NonNull ListenableFuture<ProfileAndCredential> getPipeRetrievalFuture(@NonNull SignalServiceAddress address,
-  private static @NonNull ProfileAndCredential retrieveProfileInternal(@NonNull SignalServiceAddress address,
+                                                                                        @NonNull Optional<ProfileKey> profileKey,
-                                                                       @NonNull Optional<ProfileKey> profileKey,
+                                                                                        @NonNull Optional<UnidentifiedAccess> unidentifiedAccess,
-                                                                       @NonNull Optional<UnidentifiedAccess> unidentifiedAccess,
+                                                                                        @NonNull SignalServiceProfile.RequestType requestType)
-                                                                       @NonNull SignalServiceProfile.RequestType requestType)
+      throws IOException
-    throws IOException
   {
     SignalServiceMessagePipe authPipe         = IncomingMessageObserver.getPipe();
     SignalServiceMessagePipe unidentifiedPipe = IncomingMessageObserver.getUnidentifiedPipe();
     SignalServiceMessagePipe pipe             = unidentifiedPipe != null && unidentifiedAccess.isPresent() ? unidentifiedPipe
                                                                                                            : authPipe;
-
     if (pipe != null) {
-      try {
+      return pipe.getProfile(address, profileKey, unidentifiedAccess, requestType);
-        return pipe.getProfile(address, profileKey, unidentifiedAccess, requestType);
-      } catch (IOException e) {
-        Log.w(TAG, "Websocket request failed. Falling back to REST.", e);
-      }
     }
 
+    throw new IOException("No pipe available!");
+  }
+
+  private static @NonNull ListenableFuture<ProfileAndCredential> getSocketRetrievalFuture(@NonNull SignalServiceAddress address,
+                                                                                          @NonNull Optional<ProfileKey> profileKey,
+                                                                                          @NonNull Optional<UnidentifiedAccess> unidentifiedAccess,
+                                                                                          @NonNull SignalServiceProfile.RequestType requestType)
+  {
     SignalServiceMessageReceiver receiver = ApplicationDependencies.getSignalServiceMessageReceiver();
-    try {
+    return receiver.retrieveProfile(address, profileKey, unidentifiedAccess, requestType);
-      return receiver.retrieveProfile(address, profileKey, unidentifiedAccess, requestType);
-    } catch (VerificationFailedException e) {
-      throw new IOException("Verification Problem", e);
-    }
   }
 
   private static Optional<UnidentifiedAccess> getUnidentifiedAccess(@NonNull Context context, @NonNull Recipient recipient) {

app/src/test/java/org/thoughtcrime/securesms/jobmanager/migrations/RecipientIdJobMigrationTest.java

@@ -145,21 +145,6 @@ public class RecipientIdJobMigrationTest {
     new MultiDeviceVerifiedUpdateJob.Factory().create(mock(Job.Parameters.class), converted.getData());
   }
 
-  @Test
-  public void migrate_retrieveProfileJob() throws Exception {
-    JobData testData = new JobData("RetrieveProfileJob", null, new Data.Builder().putString("address", "+16101234567").build());
-    mockRecipientResolve("+16101234567", 1);
-
-    RecipientIdJobMigration subject   = new RecipientIdJobMigration(mock(Application.class));
-    JobData                 converted = subject.migrate(testData);
-
-    assertEquals("RetrieveProfileJob", converted.getFactoryKey());
-    assertNull(converted.getQueueKey());
-    assertEquals("1", converted.getData().getString("recipient"));
-
-    new RetrieveProfileJob.Factory().create(mock(Job.Parameters.class), converted.getData());
-  }
-
   @Test
   public void migrate_pushGroupSendJob_null() throws Exception {
     JobData testData = new JobData("PushGroupSendJob", "someGroupId", new Data.Builder().putString("filter_address", null)

app/src/test/java/org/thoughtcrime/securesms/profiles/ProfileNameTest.java

@@ -140,7 +140,7 @@ public final class ProfileNameTest {
     String data = name.serialize();
 
     // THEN
-    assertEquals(data, "Given\0");
+    assertEquals(data, "Given");
   }
 
   @Test

libsignal/service/src/main/java/org/whispersystems/signalservice/api/SignalServiceAccountManager.java

@@ -30,6 +30,7 @@ import org.whispersystems.signalservice.api.groupsv2.GroupsV2Operations;
 import org.whispersystems.signalservice.api.kbs.MasterKey;
 import org.whispersystems.signalservice.api.messages.calls.TurnServerInfo;
 import org.whispersystems.signalservice.api.messages.multidevice.DeviceInfo;
+import org.whispersystems.signalservice.api.profiles.ProfileAndCredential;
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfile;
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfileWrite;
 import org.whispersystems.signalservice.api.push.ContactTokenDetails;
@@ -78,6 +79,7 @@ import java.security.KeyStore;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.security.SignatureException;
+import java.sql.Time;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -89,6 +91,9 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static org.whispersystems.signalservice.internal.push.ProvisioningProtos.ProvisionMessage;
 import static org.whispersystems.signalservice.internal.push.ProvisioningProtos.ProvisioningVersion;
@@ -705,9 +710,22 @@ public class SignalServiceAccountManager {
   }
 
   public Optional<ProfileKeyCredential> resolveProfileKeyCredential(UUID uuid, ProfileKey profileKey)
-      throws NonSuccessfulResponseCodeException, PushNetworkException, VerificationFailedException
+      throws NonSuccessfulResponseCodeException, PushNetworkException
   {
-    return this.pushServiceSocket.retrieveVersionedProfileAndCredential(uuid, profileKey, Optional.absent()).getProfileKeyCredential();
+    try {
+      ProfileAndCredential credential = this.pushServiceSocket.retrieveVersionedProfileAndCredential(uuid, profileKey, Optional.absent()).get(10, TimeUnit.SECONDS);
+      return credential.getProfileKeyCredential();
+    } catch (InterruptedException | TimeoutException e) {
+      throw new PushNetworkException(e);
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof NonSuccessfulResponseCodeException) {
+        throw (NonSuccessfulResponseCodeException) e.getCause();
+      } else if (e.getCause() instanceof PushNetworkException) {
+        throw (PushNetworkException) e.getCause();
+      } else {
+        throw new PushNetworkException(e);
+      }
+    }
   }
 
   public void setUsername(String username) throws IOException {

libsignal/service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessagePipe.java

@@ -23,6 +23,8 @@ import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope;
 import org.whispersystems.signalservice.api.profiles.ProfileAndCredential;
 import org.whispersystems.signalservice.api.profiles.SignalServiceProfile;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
+import org.whispersystems.signalservice.api.push.exceptions.NonSuccessfulResponseCodeException;
+import org.whispersystems.signalservice.api.push.exceptions.NotFoundException;
 import org.whispersystems.signalservice.api.util.CredentialsProvider;
 import org.whispersystems.signalservice.internal.push.AttachmentV2UploadAttributes;
 import org.whispersystems.signalservice.internal.push.AttachmentV3UploadAttributes;
@@ -190,64 +192,63 @@ public class SignalServiceMessagePipe {
     });
   }
 
-  public ProfileAndCredential getProfile(SignalServiceAddress address,
+  public ListenableFuture<ProfileAndCredential> getProfile(SignalServiceAddress address,
-                                         Optional<ProfileKey> profileKey,
+                                                           Optional<ProfileKey> profileKey,
-                                         Optional<UnidentifiedAccess> unidentifiedAccess,
+                                                           Optional<UnidentifiedAccess> unidentifiedAccess,
-                                         SignalServiceProfile.RequestType requestType)
+                                                           SignalServiceProfile.RequestType requestType)
       throws IOException
   {
-    try {
+    List<String> headers = new LinkedList<>();
-      List<String> headers = new LinkedList<>();
 
-      if (unidentifiedAccess.isPresent()) {
+    if (unidentifiedAccess.isPresent()) {
-        headers.add("Unidentified-Access-Key:" + Base64.encodeBytes(unidentifiedAccess.get().getUnidentifiedAccessKey()));
+      headers.add("Unidentified-Access-Key:" + Base64.encodeBytes(unidentifiedAccess.get().getUnidentifiedAccessKey()));
-      }
+    }
 
-      Optional<UUID>                     uuid           = address.getUuid();
+    Optional<UUID>                     uuid           = address.getUuid();
-      SecureRandom                       random         = new SecureRandom();
+    SecureRandom                       random         = new SecureRandom();
-      ProfileKeyCredentialRequestContext requestContext = null;
+    ProfileKeyCredentialRequestContext requestContext = null;
 
-      WebSocketRequestMessage.Builder builder = WebSocketRequestMessage.newBuilder()
+    WebSocketRequestMessage.Builder builder = WebSocketRequestMessage.newBuilder()
-                                                                       .setId(random.nextLong())
+                                                                     .setId(random.nextLong())
-                                                                       .setVerb("GET")
+                                                                     .setVerb("GET")
-                                                                       .addAllHeaders(headers);
+                                                                     .addAllHeaders(headers);
 
-      if (uuid.isPresent() && profileKey.isPresent()) {
+    if (uuid.isPresent() && profileKey.isPresent()) {
-        UUID              target               = uuid.get();
+      UUID              target               = uuid.get();
-        ProfileKeyVersion profileKeyIdentifier = profileKey.get().getProfileKeyVersion(target);
+      ProfileKeyVersion profileKeyIdentifier = profileKey.get().getProfileKeyVersion(target);
-        String            version              = profileKeyIdentifier.serialize();
+      String            version              = profileKeyIdentifier.serialize();
 
-        if (requestType == SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL) {
+      if (requestType == SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL) {
-          requestContext = clientZkProfile.createProfileKeyCredentialRequestContext(random, target, profileKey.get());
+        requestContext = clientZkProfile.createProfileKeyCredentialRequestContext(random, target, profileKey.get());
 
-          ProfileKeyCredentialRequest request           = requestContext.getRequest();
+        ProfileKeyCredentialRequest request           = requestContext.getRequest();
-          String                      credentialRequest = Hex.toStringCondensed(request.serialize());
+        String                      credentialRequest = Hex.toStringCondensed(request.serialize());
 
-          builder.setPath(String.format("/v1/profile/%s/%s/%s", target, version, credentialRequest));
+        builder.setPath(String.format("/v1/profile/%s/%s/%s", target, version, credentialRequest));
-        } else {
-          builder.setPath(String.format("/v1/profile/%s/%s", target, version));
-        }
       } else {
-        builder.setPath(String.format("/v1/profile/%s", address.getIdentifier()));
+        builder.setPath(String.format("/v1/profile/%s/%s", target, version));
       }
+    } else {
+      builder.setPath(String.format("/v1/profile/%s", address.getIdentifier()));
+    }
 
-      WebSocketRequestMessage requestMessage = builder.build();
+    final ProfileKeyCredentialRequestContext finalRequestContext = requestContext;
+    WebSocketRequestMessage requestMessage = builder.build();
 
-      WebsocketResponse response = websocket.sendRequest(requestMessage).get(10, TimeUnit.SECONDS);
+    return FutureTransformers.map(websocket.sendRequest(requestMessage), response -> {
-
+      if (response.getStatus() == 404) {
-      if (response.getStatus() < 200 || response.getStatus() >= 300) {
+        throw new NotFoundException("Not found");
-        throw new IOException("Non-successful response: " + response.getStatus());
+      } else if (response.getStatus() < 200 || response.getStatus() >= 300) {
+        throw new NonSuccessfulResponseCodeException("Non-successful response: " + response.getStatus());
       }
 
       SignalServiceProfile signalServiceProfile = JsonUtil.fromJson(response.getBody(), SignalServiceProfile.class);
-      ProfileKeyCredential profileKeyCredential = requestContext != null && signalServiceProfile.getProfileKeyCredentialResponse() != null
+      ProfileKeyCredential profileKeyCredential = finalRequestContext != null && signalServiceProfile.getProfileKeyCredentialResponse() != null
-                                                  ? clientZkProfile.receiveProfileKeyCredential(requestContext, signalServiceProfile.getProfileKeyCredentialResponse())
+                                                    ? clientZkProfile.receiveProfileKeyCredential(finalRequestContext, signalServiceProfile.getProfileKeyCredentialResponse())
-                                                  : null;
+                                                    : null;
 
       return new ProfileAndCredential(signalServiceProfile, requestType, Optional.fromNullable(profileKeyCredential));
-    } catch (InterruptedException | ExecutionException | TimeoutException | VerificationFailedException e) {
+    });
-      throw new IOException(e);
-    }
   }
 
   public AttachmentV2UploadAttributes getAttachmentV2UploadAttributes() throws IOException {

libsignal/service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageReceiver.java

@@ -35,6 +35,8 @@ import org.whispersystems.signalservice.internal.push.SignalServiceEnvelopeEntit
 import org.whispersystems.signalservice.internal.sticker.StickerProtos;
 import org.whispersystems.signalservice.internal.util.StaticCredentialsProvider;
 import org.whispersystems.signalservice.internal.util.Util;
+import org.whispersystems.signalservice.internal.util.concurrent.FutureTransformers;
+import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
 import org.whispersystems.signalservice.internal.websocket.WebSocketConnection;
 
 import java.io.ByteArrayOutputStream;
@@ -120,11 +122,10 @@ public class SignalServiceMessageReceiver {
     return retrieveAttachment(pointer, destination, maxSizeBytes, null);
   }
 
-  public ProfileAndCredential retrieveProfile(SignalServiceAddress address,
+  public ListenableFuture<ProfileAndCredential> retrieveProfile(SignalServiceAddress address,
-                                              Optional<ProfileKey> profileKey,
+                                                                 Optional<ProfileKey> profileKey,
-                                              Optional<UnidentifiedAccess> unidentifiedAccess,
+                                                                 Optional<UnidentifiedAccess> unidentifiedAccess,
-                                              SignalServiceProfile.RequestType requestType)
+                                                                 SignalServiceProfile.RequestType requestType)
-      throws NonSuccessfulResponseCodeException, PushNetworkException, VerificationFailedException
   {
     Optional<UUID> uuid = address.getUuid();
 
@@ -132,14 +133,18 @@ public class SignalServiceMessageReceiver {
       if (requestType == SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL) {
         return socket.retrieveVersionedProfileAndCredential(uuid.get(), profileKey.get(), unidentifiedAccess);
       } else {
-        return new ProfileAndCredential(socket.retrieveVersionedProfile(uuid.get(), profileKey.get(), unidentifiedAccess),
+        return FutureTransformers.map(socket.retrieveVersionedProfile(uuid.get(), profileKey.get(), unidentifiedAccess), profile -> {
-                                        SignalServiceProfile.RequestType.PROFILE,
+          return new ProfileAndCredential(profile,
-                                        Optional.absent());
+                                          SignalServiceProfile.RequestType.PROFILE,
+                                          Optional.absent());
+        });
       }
     } else {
-      return new ProfileAndCredential(socket.retrieveProfile(address, unidentifiedAccess),
+      return FutureTransformers.map(socket.retrieveProfile(address, unidentifiedAccess), profile -> {
-                                      SignalServiceProfile.RequestType.PROFILE,
+        return new ProfileAndCredential(profile,
-                                      Optional.absent());
+                                        SignalServiceProfile.RequestType.PROFILE,
+                                        Optional.absent());
+      });
     }
   }
 

libsignal/service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java

@@ -1226,7 +1226,7 @@ public class SignalServiceMessageSender {
           Log.w(TAG, e);
           results.add(SendMessageResult.identityFailure(recipient, ((UntrustedIdentityException) e.getCause()).getIdentityKey()));
         } else if (e.getCause() instanceof UnregisteredUserException) {
-          Log.w(TAG, e);
+          Log.w(TAG, "Found unregistered user.");
           results.add(SendMessageResult.unregisteredFailure(recipient));
         } else if (e.getCause() instanceof PushNetworkException) {
           Log.w(TAG, e);

libsignal/service/src/main/java/org/whispersystems/signalservice/internal/push/PushServiceSocket.java

@@ -90,6 +90,9 @@ import org.whispersystems.signalservice.internal.util.BlacklistingTrustManager;
 import org.whispersystems.signalservice.internal.util.Hex;
 import org.whispersystems.signalservice.internal.util.JsonUtil;
 import org.whispersystems.signalservice.internal.util.Util;
+import org.whispersystems.signalservice.internal.util.concurrent.FutureTransformers;
+import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
+import org.whispersystems.signalservice.internal.util.concurrent.SettableFuture;
 import org.whispersystems.util.Base64;
 
 import java.io.ByteArrayInputStream;
@@ -113,6 +116,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.SSLContext;
@@ -120,6 +124,7 @@ import javax.net.ssl.TrustManager;
 import javax.net.ssl.X509TrustManager;
 
 import okhttp3.Call;
+import okhttp3.Callback;
 import okhttp3.ConnectionSpec;
 import okhttp3.Credentials;
 import okhttp3.Dns;
@@ -280,9 +285,9 @@ public class PushServiceSocket {
   }
 
   public VerifyAccountResponse verifyAccountCode(String verificationCode, String signalingKey, int registrationId, boolean fetchesMessages,
-                                String pin, String registrationLock,
+                                                 String pin, String registrationLock,
-                                byte[] unidentifiedAccessKey, boolean unrestrictedUnidentifiedAccess,
+                                                 byte[] unidentifiedAccessKey, boolean unrestrictedUnidentifiedAccess,
-                                SignalServiceProfile.Capabilities capabilities)
+                                                 SignalServiceProfile.Capabilities capabilities)
       throws IOException
   {
     AccountAttributes signalingKeyEntity = new AccountAttributes(signalingKey, registrationId, fetchesMessages, pin, registrationLock, unidentifiedAccessKey, unrestrictedUnidentifiedAccess, capabilities);
@@ -376,6 +381,15 @@ public class PushServiceSocket {
     }
   }
 
+  public Future<SendMessageResponse> submitMessage(OutgoingPushMessageList bundle, Optional<UnidentifiedAccess> unidentifiedAccess) {
+    ListenableFuture<String> response = submitServiceRequest(String.format(MESSAGE_PATH, bundle.getDestination()), "PUT", JsonUtil.toJson(bundle), NO_HEADERS, unidentifiedAccess);
+
+    return FutureTransformers.map(response, body -> {
+      return body == null ? new SendMessageResponse(false)
+          : JsonUtil.fromJson(body, SendMessageResponse.class);
+    });
+  }
+
   public List<SignalServiceEnvelopeEntity> getMessages() throws IOException {
     String responseText = makeServiceRequest(String.format(MESSAGE_PATH, ""), "GET", null);
     return JsonUtil.fromJson(responseText, SignalServiceEnvelopeEntityList.class).getMessages();
@@ -398,17 +412,17 @@ public class PushServiceSocket {
 
     for (PreKeyRecord record : records) {
       PreKeyEntity entity = new PreKeyEntity(record.getId(),
-                                             record.getKeyPair().getPublicKey());
+          record.getKeyPair().getPublicKey());
 
       entities.add(entity);
     }
 
     SignedPreKeyEntity signedPreKeyEntity = new SignedPreKeyEntity(signedPreKey.getId(),
-                                                                   signedPreKey.getKeyPair().getPublicKey(),
+        signedPreKey.getKeyPair().getPublicKey(),
-                                                                   signedPreKey.getSignature());
+        signedPreKey.getSignature());
 
     makeServiceRequest(String.format(PREKEY_PATH, ""), "PUT",
-                       JsonUtil.toJson(new PreKeyState(entities, signedPreKeyEntity, identityKey)));
+        JsonUtil.toJson(new PreKeyState(entities, signedPreKeyEntity, identityKey)));
   }
 
   public int getAvailablePreKeys() throws IOException {
@@ -458,8 +472,8 @@ public class PushServiceSocket {
         }
 
         bundles.add(new PreKeyBundle(device.getRegistrationId(), device.getDeviceId(), preKeyId,
-                                     preKey, signedPreKeyId, signedPreKey, signedPreKeySignature,
+            preKey, signedPreKeyId, signedPreKey, signedPreKeySignature,
-                                     response.getIdentityKey()));
+            response.getIdentityKey()));
       }
 
       return bundles;
@@ -569,17 +583,17 @@ public class PushServiceSocket {
     return output.toByteArray();
   }
 
-  public SignalServiceProfile retrieveProfile(SignalServiceAddress target, Optional<UnidentifiedAccess> unidentifiedAccess)
+  public ListenableFuture<SignalServiceProfile> retrieveProfile(SignalServiceAddress target, Optional<UnidentifiedAccess> unidentifiedAccess) {
-      throws NonSuccessfulResponseCodeException, PushNetworkException
+    ListenableFuture<String> response = submitServiceRequest(String.format(PROFILE_PATH, target.getIdentifier()), "GET", null, NO_HEADERS, unidentifiedAccess);
-  {
-    String response = makeServiceRequest(String.format(PROFILE_PATH, target.getIdentifier()), "GET", null, NO_HEADERS, unidentifiedAccess);
 
-    try {
+    return FutureTransformers.map(response, body -> {
-      return JsonUtil.fromJson(response, SignalServiceProfile.class);
+      try {
-    } catch (IOException e) {
+        return JsonUtil.fromJson(body, SignalServiceProfile.class);
-      Log.w(TAG, e);
+      } catch (IOException e) {
-      throw new NonSuccessfulResponseCodeException("Unable to parse entity");
+        Log.w(TAG, e);
-    }
+        throw new NonSuccessfulResponseCodeException("Unable to parse entity");
+      }
+    });
   }
 
   public SignalServiceProfile retrieveProfileByUsername(String username, Optional<UnidentifiedAccess> unidentifiedAccess)
@@ -595,9 +609,7 @@ public class PushServiceSocket {
     }
   }
 
-  public ProfileAndCredential retrieveVersionedProfileAndCredential(UUID target, ProfileKey profileKey, Optional<UnidentifiedAccess> unidentifiedAccess)
+  public ListenableFuture<ProfileAndCredential> retrieveVersionedProfileAndCredential(UUID target, ProfileKey profileKey, Optional<UnidentifiedAccess> unidentifiedAccess) {
-      throws NonSuccessfulResponseCodeException, PushNetworkException, VerificationFailedException
-  {
     ProfileKeyVersion                  profileKeyIdentifier = profileKey.getProfileKeyVersion(target);
     ProfileKeyCredentialRequestContext requestContext       = clientZkProfileOperations.createProfileKeyCredentialRequestContext(random, target, profileKey);
     ProfileKeyCredentialRequest        request              = requestContext.getRequest();
@@ -606,38 +618,47 @@ public class PushServiceSocket {
     String credentialRequest = Hex.toStringCondensed(request.serialize());
     String subPath           = String.format("%s/%s/%s", target, version, credentialRequest);
 
-    String response = makeServiceRequest(String.format(PROFILE_PATH, subPath), "GET", null, NO_HEADERS, unidentifiedAccess);
+    ListenableFuture<String> response = submitServiceRequest(String.format(PROFILE_PATH, subPath), "GET", null, NO_HEADERS, unidentifiedAccess);
 
+    return FutureTransformers.map(response, body -> formatProfileAndCredentialBody(requestContext, body));
+  }
+
+  private ProfileAndCredential formatProfileAndCredentialBody(ProfileKeyCredentialRequestContext requestContext, String body)
+      throws NonSuccessfulResponseCodeException
+  {
     try {
-      SignalServiceProfile signalServiceProfile = JsonUtil.fromJson(response, SignalServiceProfile.class);
+      SignalServiceProfile signalServiceProfile = JsonUtil.fromJson(body, SignalServiceProfile.class);
 
-      ProfileKeyCredential profileKeyCredential = signalServiceProfile.getProfileKeyCredentialResponse() != null
+      try {
-                                                ? clientZkProfileOperations.receiveProfileKeyCredential(requestContext, signalServiceProfile.getProfileKeyCredentialResponse())
+        ProfileKeyCredential profileKeyCredential = signalServiceProfile.getProfileKeyCredentialResponse() != null
-                                                : null;
+                                                      ? clientZkProfileOperations.receiveProfileKeyCredential(requestContext, signalServiceProfile.getProfileKeyCredentialResponse())
-
+                                                      : null;
-      return new ProfileAndCredential(signalServiceProfile, SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL, Optional.fromNullable(profileKeyCredential));
+        return new ProfileAndCredential(signalServiceProfile, SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL, Optional.fromNullable(profileKeyCredential));
+      } catch (VerificationFailedException e) {
+        Log.w(TAG, "Failed to verify credential.", e);
+        return new ProfileAndCredential(signalServiceProfile, SignalServiceProfile.RequestType.PROFILE_AND_CREDENTIAL, Optional.absent());
+      }
     } catch (IOException e) {
       Log.w(TAG, e);
       throw new NonSuccessfulResponseCodeException("Unable to parse entity");
     }
   }
 
-  public SignalServiceProfile retrieveVersionedProfile(UUID target, ProfileKey profileKey, Optional<UnidentifiedAccess> unidentifiedAccess)
+  public ListenableFuture<SignalServiceProfile> retrieveVersionedProfile(UUID target, ProfileKey profileKey, Optional<UnidentifiedAccess> unidentifiedAccess) {
-      throws NonSuccessfulResponseCodeException, PushNetworkException
-  {
     ProfileKeyVersion profileKeyIdentifier = profileKey.getProfileKeyVersion(target);
 
-    String version = profileKeyIdentifier.serialize();
+    String                   version  = profileKeyIdentifier.serialize();
-    String subPath = String.format("%s/%s", target, version);
+    String                   subPath  = String.format("%s/%s", target, version);
+    ListenableFuture<String> response = submitServiceRequest(String.format(PROFILE_PATH, subPath), "GET", null, NO_HEADERS, unidentifiedAccess);
 
-    String response = makeServiceRequest(String.format(PROFILE_PATH, subPath), "GET", null, NO_HEADERS, unidentifiedAccess);
+    return FutureTransformers.map(response, body -> {
-
+      try {
-    try {
+        return JsonUtil.fromJson(body, SignalServiceProfile.class);
-      return JsonUtil.fromJson(response, SignalServiceProfile.class);
+      } catch (IOException e) {
-    } catch (IOException e) {
+        Log.w(TAG, e);
-      Log.w(TAG, e);
+        throw new NonSuccessfulResponseCodeException("Unable to parse entity");
-      throw new NonSuccessfulResponseCodeException("Unable to parse entity");
+      }
-    }
+    });
   }
 
   public void retrieveProfileAvatar(String path, File destination, long maxSizeBytes)
@@ -700,7 +721,7 @@ public class PushServiceSocket {
     String response = makeServiceRequest(String.format(PROFILE_PATH, ""), "PUT", requestBody);
 
     if (signalServiceProfileWrite.hasAvatar() && profileAvatar != null) {
-       try {
+      try {
         formAttributes = JsonUtil.fromJson(response, ProfileAvatarUploadAttributes.class);
       } catch (IOException e) {
         Log.w(TAG, e);
@@ -1323,14 +1344,48 @@ public class PushServiceSocket {
 
   private static RequestBody jsonRequestBody(String jsonBody) {
     return jsonBody != null
-           ? RequestBody.create(MediaType.parse("application/json"), jsonBody)
+        ? RequestBody.create(MediaType.parse("application/json"), jsonBody)
-           : null;
+        : null;
   }
 
   private static RequestBody protobufRequestBody(MessageLite protobufBody) {
     return protobufBody != null
-           ? RequestBody.create(MediaType.parse("application/x-protobuf"), protobufBody.toByteArray())
+        ? RequestBody.create(MediaType.parse("application/x-protobuf"), protobufBody.toByteArray())
-           : null;
+        : null;
+  }
+
+
+  private ListenableFuture<String> submitServiceRequest(String urlFragment, String method, String jsonBody, Map<String, String> headers, Optional<UnidentifiedAccess> unidentifiedAccessKey) {
+    OkHttpClient okHttpClient = buildOkHttpClient(unidentifiedAccessKey.isPresent());
+    Call         call         = okHttpClient.newCall(buildServiceRequest(urlFragment, method, jsonRequestBody(jsonBody), headers, unidentifiedAccessKey));
+
+    synchronized (connections) {
+      connections.add(call);
+    }
+
+    SettableFuture<String> bodyFuture = new SettableFuture<>();
+
+    call.enqueue(new Callback() {
+      @Override
+      public void onResponse(Call call, Response response) {
+        try (ResponseBody body = validateServiceResponse(response)) {
+          try {
+            bodyFuture.set(body.string());
+          } catch (IOException e) {
+            throw new PushNetworkException(e);
+          }
+        } catch (IOException e) {
+          bodyFuture.setException(e);
+        }
+      }
+
+      @Override
+      public void onFailure(Call call, IOException e) {
+        bodyFuture.setException(e);
+      }
+    });
+
+    return bodyFuture;
   }
 
   private ResponseBody makeServiceBodyRequest(String urlFragment,
@@ -1343,12 +1398,16 @@ public class PushServiceSocket {
   {
     Response response = getServiceConnection(urlFragment, method, body, headers, unidentifiedAccessKey);
 
+    responseCodeHandler.handle(response.code());
+
+    return validateServiceResponse(response);
+  }
+
+  private ResponseBody validateServiceResponse(Response response) throws NonSuccessfulResponseCodeException, PushNetworkException {
     int          responseCode    = response.code();
     String       responseMessage = response.message();
     ResponseBody responseBody    = response.body();
 
-    responseCodeHandler.handle(responseCode);
-
     switch (responseCode) {
       case 413:
         throw new RateLimitException("Rate limit exceeded: " + responseCode);
@@ -1427,42 +1486,8 @@ public class PushServiceSocket {
       throws PushNetworkException
   {
     try {
-      ServiceConnectionHolder connectionHolder = (ServiceConnectionHolder) getRandom(serviceClients, random);
+      OkHttpClient okHttpClient = buildOkHttpClient(unidentifiedAccess.isPresent());
-      OkHttpClient            baseClient       = unidentifiedAccess.isPresent() ? connectionHolder.getUnidentifiedClient() : connectionHolder.getClient();
+      Call         call         = okHttpClient.newCall(buildServiceRequest(urlFragment, method, body, headers, unidentifiedAccess));
-      OkHttpClient            okHttpClient     = baseClient.newBuilder()
-                                                           .connectTimeout(soTimeoutMillis, TimeUnit.MILLISECONDS)
-                                                           .readTimeout(soTimeoutMillis, TimeUnit.MILLISECONDS)
-                                                           .build();
-
-//      Log.d(TAG, "Push service URL: " + connectionHolder.getUrl());
-//      Log.d(TAG, "Opening URL: " + String.format("%s%s", connectionHolder.getUrl(), urlFragment));
-      Log.d(TAG, "Opening URL: <REDACTED>");
-
-      Request.Builder request = new Request.Builder();
-      request.url(String.format("%s%s", connectionHolder.getUrl(), urlFragment));
-      request.method(method, body);
-
-      for (Map.Entry<String, String> header : headers.entrySet()) {
-        request.addHeader(header.getKey(), header.getValue());
-      }
-
-      if (!headers.containsKey("Authorization")) {
-        if (unidentifiedAccess.isPresent()) {
-          request.addHeader("Unidentified-Access-Key", Base64.encodeBytes(unidentifiedAccess.get().getUnidentifiedAccessKey()));
-        } else if (credentialsProvider.getPassword() != null) {
-          request.addHeader("Authorization", getAuthorizationHeader(credentialsProvider));
-        }
-      }
-
-      if (signalAgent != null) {
-        request.addHeader("X-Signal-Agent", signalAgent);
-      }
-
-      if (connectionHolder.getHostHeader().isPresent()) {
-        request.addHeader("Host", connectionHolder.getHostHeader().get());
-      }
-
-      Call call = okHttpClient.newCall(request.build());
 
       synchronized (connections) {
         connections.add(call);
@@ -1480,6 +1505,51 @@ public class PushServiceSocket {
     }
   }
 
+  private OkHttpClient buildOkHttpClient(boolean unidentified) {
+    ServiceConnectionHolder connectionHolder = (ServiceConnectionHolder) getRandom(serviceClients, random);
+    OkHttpClient            baseClient       = unidentified ? connectionHolder.getUnidentifiedClient() : connectionHolder.getClient();
+
+    return baseClient.newBuilder()
+                     .connectTimeout(soTimeoutMillis, TimeUnit.MILLISECONDS)
+                     .readTimeout(soTimeoutMillis, TimeUnit.MILLISECONDS)
+                     .build();
+  }
+
+  private Request buildServiceRequest(String urlFragment, String method, RequestBody body, Map<String, String> headers, Optional<UnidentifiedAccess> unidentifiedAccess) {
+    ServiceConnectionHolder connectionHolder = (ServiceConnectionHolder) getRandom(serviceClients, random);
+
+//      Log.d(TAG, "Push service URL: " + connectionHolder.getUrl());
+//      Log.d(TAG, "Opening URL: " + String.format("%s%s", connectionHolder.getUrl(), urlFragment));
+    Log.d(TAG, "Opening URL: <REDACTED>");
+
+    Request.Builder request = new Request.Builder();
+    request.url(String.format("%s%s", connectionHolder.getUrl(), urlFragment));
+    request.method(method, body);
+
+    for (Map.Entry<String, String> header : headers.entrySet()) {
+      request.addHeader(header.getKey(), header.getValue());
+    }
+
+    if (!headers.containsKey("Authorization")) {
+      if (unidentifiedAccess.isPresent()) {
+        request.addHeader("Unidentified-Access-Key", Base64.encodeBytes(unidentifiedAccess.get().getUnidentifiedAccessKey()));
+      } else if (credentialsProvider.getPassword() != null) {
+        request.addHeader("Authorization", getAuthorizationHeader(credentialsProvider));
+      }
+    }
+
+    if (signalAgent != null) {
+      request.addHeader("X-Signal-Agent", signalAgent);
+    }
+
+    if (connectionHolder.getHostHeader().isPresent()) {
+      request.addHeader("Host", connectionHolder.getHostHeader().get());
+    }
+
+    return request.build();
+  }
+
+
   private ConnectionHolder[] clientsFor(ClientSet clientSet) {
     switch (clientSet) {
       case ContactDiscovery:

libsignal/service/src/main/java/org/whispersystems/signalservice/internal/util/concurrent/CascadingFuture.java

@@ -0,0 +1,100 @@
+package org.whispersystems.signalservice.internal.util.concurrent;
+
+import org.whispersystems.libsignal.logging.Log;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * A future that allows you to have multiple ways to compute a result. If one fails, the calculation
+ * will fall back to the next in the list.
+ *
+ * You will only see a failure if the last attempt in the list fails.
+ */
+public final class CascadingFuture<T> implements ListenableFuture<T> {
+
+  private static final String TAG = CascadingFuture.class.getSimpleName();
+
+  private SettableFuture<T> result;
+
+  public CascadingFuture(List<Callable<ListenableFuture<T>>> callables, ExceptionChecker exceptionChecker) {
+    if (callables.isEmpty()) {
+      throw new IllegalArgumentException("Must have at least one callable!");
+    }
+
+    this.result = new SettableFuture<>();
+
+    doNext(new ArrayList<>(callables), exceptionChecker);
+  }
+
+  @Override
+  public boolean cancel(boolean mayInterruptIfRunning) {
+    return result.cancel(mayInterruptIfRunning);
+  }
+
+  @Override
+  public boolean isCancelled() {
+    return result.isCancelled();
+  }
+
+  @Override
+  public boolean isDone() {
+    return result.isDone();
+  }
+
+  @Override
+  public T get() throws ExecutionException, InterruptedException {
+    return result.get();
+  }
+
+  @Override
+  public T get(long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException {
+    return result.get(timeout, unit);
+  }
+
+  @Override
+  public void addListener(Listener<T> listener) {
+    result.addListener(listener);
+  }
+
+  private void doNext(List<Callable<ListenableFuture<T>>> callables, ExceptionChecker exceptionChecker) {
+    Callable<ListenableFuture<T>> callable = callables.remove(0);
+    try {
+      ListenableFuture<T> future = callable.call();
+
+      future.addListener(new ListenableFuture.Listener<T>() {
+        @Override
+        public void onSuccess(T value) {
+          result.set(value);
+        }
+
+        @Override
+        public void onFailure(ExecutionException e) {
+          if (callables.isEmpty() || !exceptionChecker.shouldContinue(e)) {
+            Log.w(TAG, e);
+            result.setException(e.getCause());
+          } else if (!result.isCancelled()) {
+            doNext(callables, exceptionChecker);
+          }
+        }
+      });
+    } catch (Exception e) {
+      if (callables.isEmpty() || !exceptionChecker.shouldContinue(e)) {
+        result.setException(e.getCause());
+      } else if (!result.isCancelled()) {
+        Log.w(TAG, e);
+        doNext(callables, exceptionChecker);
+      }
+    }
+  }
+
+  public interface ExceptionChecker {
+    boolean shouldContinue(Exception e);
+  }
+}