Add message processing performance test.

app/build.gradle

@@ -145,7 +145,7 @@ android {
 
     packagingOptions {
         resources {
-            excludes += ['LICENSE.txt', 'LICENSE', 'NOTICE', 'asm-license.txt', 'META-INF/LICENSE', 'META-INF/NOTICE', 'META-INF/proguard/androidx-annotations.pro', 'libsignal_jni.dylib', 'signal_jni.dll']
+            excludes += ['LICENSE.txt', 'LICENSE', 'NOTICE', 'asm-license.txt', 'META-INF/LICENSE', 'META-INF/LICENSE.md', 'META-INF/NOTICE', 'META-INF/LICENSE-notice.md', 'META-INF/proguard/androidx-annotations.pro', 'libsignal_jni.dylib', 'signal_jni.dll']
         }
     }
 
@@ -570,6 +570,7 @@ dependencies {
     androidTestImplementation testLibs.androidx.test.ext.junit.ktx
     androidTestImplementation testLibs.mockito.android
     androidTestImplementation testLibs.mockito.kotlin
+    androidTestImplementation testLibs.mockk.android
     androidTestImplementation testLibs.square.okhttp.mockserver
 
     instrumentationImplementation (libs.androidx.fragment.testing) {

app/src/androidTest/java/org/thoughtcrime/securesms/SignalInstrumentationApplicationContext.kt

@@ -1,15 +1,39 @@
 package org.thoughtcrime.securesms
 
+import org.signal.core.util.concurrent.SignalExecutors
+import org.signal.core.util.logging.AndroidLogger
+import org.signal.core.util.logging.Log
+import org.signal.libsignal.protocol.logging.SignalProtocolLoggerProvider
+import org.thoughtcrime.securesms.database.LogDatabase
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencyProvider
 import org.thoughtcrime.securesms.dependencies.InstrumentationApplicationDependencyProvider
+import org.thoughtcrime.securesms.logging.CustomSignalProtocolLogger
+import org.thoughtcrime.securesms.logging.PersistentLogger
+import org.thoughtcrime.securesms.testing.InMemoryLogger
 
 /**
  * Application context for running instrumentation tests (aka androidTests).
  */
 class SignalInstrumentationApplicationContext : ApplicationContext() {
+
+  val inMemoryLogger: InMemoryLogger = InMemoryLogger()
+
   override fun initializeAppDependencies() {
     val default = ApplicationDependencyProvider(this)
     ApplicationDependencies.init(this, InstrumentationApplicationDependencyProvider(this, default))
   }
+
+  override fun initializeLogging() {
+    persistentLogger = PersistentLogger(this)
+
+    Log.initialize({ true }, AndroidLogger(), persistentLogger, inMemoryLogger)
+
+    SignalProtocolLoggerProvider.setProvider(CustomSignalProtocolLogger())
+
+    SignalExecutors.UNBOUNDED.execute {
+      Log.blockUntilAllWritesFinished()
+      LogDatabase.getInstance(this).trimToSize()
+    }
+  }
 }

app/src/androidTest/java/org/thoughtcrime/securesms/database/AttachmentTableTest.kt

@@ -2,6 +2,7 @@ package org.thoughtcrime.securesms.database
 
 import android.net.Uri
 import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.filters.FlakyTest
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotEquals
 import org.junit.Before
@@ -34,6 +35,7 @@ class AttachmentTableTest {
     assertEquals(attachment2.fileName, attachment.fileName)
   }
 
+  @FlakyTest
   @Test
   fun givenABlobAndDifferentTransformQuality_whenIInsert2AttachmentsForPreUpload_thenIExpectDifferentFileInfos() {
     val blob = BlobProvider.getInstance().forData(byteArrayOf(1, 2, 3, 4, 5)).createForSingleSessionInMemory()
@@ -61,6 +63,7 @@ class AttachmentTableTest {
     assertNotEquals(attachment1Info, attachment2Info)
   }
 
+  @FlakyTest
   @Test
   fun givenIdenticalAttachmentsInsertedForPreUpload_whenIUpdateAttachmentDataAndSpecifyOnlyModifyThisAttachment_thenIExpectDifferentFileInfos() {
     val blob = BlobProvider.getInstance().forData(byteArrayOf(1, 2, 3, 4, 5)).createForSingleSessionInMemory()

app/src/androidTest/java/org/thoughtcrime/securesms/messages/MessageProcessingPerformanceTest.kt

@@ -0,0 +1,188 @@
+package org.thoughtcrime.securesms.messages
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import io.mockk.every
+import io.mockk.mockkStatic
+import io.mockk.unmockkStatic
+import org.junit.After
+import org.junit.Before
+import org.junit.Ignore
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.signal.core.util.logging.Log
+import org.signal.libsignal.protocol.ecc.Curve
+import org.signal.libsignal.protocol.ecc.ECKeyPair
+import org.signal.libsignal.zkgroup.profiles.ProfileKey
+import org.thoughtcrime.securesms.crypto.UnidentifiedAccessUtil
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.testing.AliceClient
+import org.thoughtcrime.securesms.testing.BobClient
+import org.thoughtcrime.securesms.testing.Entry
+import org.thoughtcrime.securesms.testing.FakeClientHelpers
+import org.thoughtcrime.securesms.testing.SignalActivityRule
+import org.thoughtcrime.securesms.testing.awaitFor
+import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * Sends N messages from Bob to Alice to track performance of Alice's processing of messages.
+ */
+@Ignore("Ignore test in normal testing as it's a performance test with no assertions")
+@RunWith(AndroidJUnit4::class)
+class MessageProcessingPerformanceTest {
+
+  companion object {
+    private val TAG = Log.tag(MessageProcessingPerformanceTest::class.java)
+    private val TIMING_TAG = "TIMING_$TAG".substring(0..23)
+
+    private val jobFinishRegex = "\\[JOB::[a-f\\d]{8}-[a-f\\d]{4}-[a-f\\d]{4}-[a-f\\d]{4}-[a-f\\d]{12}]\\[([^]]*)]\\[\\d+] Job finished with result SUCCESS in (\\d+) ms. \\(Time Since Submission: (\\d+) ms.*".toRegex()
+  }
+
+  @get:Rule
+  val harness = SignalActivityRule()
+
+  private val trustRoot: ECKeyPair = Curve.generateKeyPair()
+
+  @Before
+  fun setup() {
+    mockkStatic(UnidentifiedAccessUtil::class)
+    every { UnidentifiedAccessUtil.getCertificateValidator() } returns FakeClientHelpers.noOpCertificateValidator
+
+    mockkStatic(MessageContentProcessor::class)
+    every { MessageContentProcessor.create(harness.application) } returns TimingMessageContentProcessor(harness.application)
+  }
+
+  @After
+  fun after() {
+    unmockkStatic(UnidentifiedAccessUtil::class)
+    unmockkStatic(MessageContentProcessor::class)
+  }
+
+  @Test
+  fun testPerformance() {
+    val aliceClient = AliceClient(
+      serviceId = harness.self.requireServiceId(),
+      e164 = harness.self.requireE164(),
+      trustRoot = trustRoot
+    )
+
+    val bob = Recipient.resolved(harness.others[0])
+    val bobClient = BobClient(
+      serviceId = bob.requireServiceId(),
+      e164 = bob.requireE164(),
+      identityKeyPair = harness.othersKeys[0],
+      trustRoot = trustRoot,
+      profileKey = ProfileKey(bob.profileKey)
+    )
+
+    // Send message from Bob to Alice (self)
+
+    val firstPreKeyMessageTimestamp = System.currentTimeMillis()
+    val encryptedEnvelope = bobClient.encrypt(firstPreKeyMessageTimestamp)
+
+    val aliceProcessFirstMessageLatch = harness
+      .inMemoryLogger
+      .getLockForUntil(TimingMessageContentProcessor.endTagPredicate(firstPreKeyMessageTimestamp))
+
+    Thread { aliceClient.process(encryptedEnvelope) }.start()
+    aliceProcessFirstMessageLatch.awaitFor(15.seconds)
+
+    // Send message from Alice to Bob
+    bobClient.decrypt(aliceClient.encrypt(System.currentTimeMillis(), bob))
+
+    // Build N messages from Bob to Alice
+
+    val messageCount = 100
+    val envelopes = ArrayList<SignalServiceEnvelope>(messageCount)
+    var now = System.currentTimeMillis()
+    for (i in 0..messageCount) {
+      envelopes += bobClient.encrypt(now)
+      now += 3
+    }
+
+    val firstTimestamp = envelopes.first().timestamp
+    val lastTimestamp = envelopes.last().timestamp
+
+    // Alice processes N messages
+
+    val aliceProcessLastMessageLatch = harness
+      .inMemoryLogger
+      .getLockForUntil(TimingMessageContentProcessor.endTagPredicate(lastTimestamp))
+
+    Thread {
+      for (envelope in envelopes) {
+        Log.i(TIMING_TAG, "Retrieved envelope! ${envelope.timestamp}")
+        aliceClient.process(envelope)
+      }
+    }.start()
+
+    // Wait for Alice to finish processing messages
+    aliceProcessLastMessageLatch.awaitFor(1.minutes)
+    harness.inMemoryLogger.flush()
+
+    // Process logs for timing data
+    val entries = harness.inMemoryLogger.entries()
+
+    // Calculate decrypt jobs
+    var skipFirst = true
+    var decryptJobCount = 0L
+    var decryptJobDuration = 0L
+    var decryptJobSinceSubmission = 0L
+    var firstDuration = 0L
+    var firstSinceSubmission = 0L
+    entries.filter { it.tag == "JobRunner" }
+      .forEach {
+        val match = jobFinishRegex.matchEntire(it.message!!)
+
+        if (match != null) {
+          val job = match.groupValues[1]
+          if (job == "PushDecryptMessageJob") {
+            if (skipFirst) {
+              skipFirst = false
+            } else {
+              val duration = match.groupValues[2].toLong()
+              val sinceSubmission = match.groupValues[3].toLong()
+              decryptJobCount++
+              decryptJobDuration += duration
+              decryptJobSinceSubmission += sinceSubmission
+
+              if (decryptJobCount == 1L) {
+                firstDuration = duration
+                firstSinceSubmission = sinceSubmission
+              }
+            }
+          }
+        }
+      }
+
+    android.util.Log.w(TAG, "Push Decrypt Job: First runtime ${firstDuration}ms First since submission: ${firstSinceSubmission}ms")
+    android.util.Log.w(TAG, "Push Decrypt Job: Average runtime: ${decryptJobDuration.toFloat() / decryptJobCount.toFloat()}ms Average since submission: ${decryptJobSinceSubmission.toFloat() / decryptJobCount.toFloat()}ms")
+
+    // Calculate MessageContentProcessor
+
+    val takeLast: List<Entry> = entries.filter { it.tag == TimingMessageContentProcessor.TAG }.drop(2)
+    val iterator = takeLast.iterator()
+    var processCount = 0L
+    var processDuration = 0L
+    while (iterator.hasNext()) {
+      val start = iterator.next()
+      val end = iterator.next()
+      processCount++
+      processDuration += end.timestamp - start.timestamp
+    }
+
+    android.util.Log.w(TAG, "MessageContentProcessor.process: Average runtime: ${processDuration.toFloat() / processCount.toFloat()}ms")
+
+    // Calculate messages per second from "retrieving" first message post session initialization to processing last message
+
+    val start = entries.first { it.message == "Retrieved envelope! $firstTimestamp" }
+    val end = entries.first { it.message == TimingMessageContentProcessor.endTag(lastTimestamp) }
+
+    val duration = (end.timestamp - start.timestamp).toFloat() / 1000f
+    val messagePerSecond = messageCount.toFloat() / duration
+
+    android.util.Log.w(TAG, "Processing $messageCount messages took ${duration}s or ${messagePerSecond}m/s")
+  }
+}

app/src/androidTest/java/org/thoughtcrime/securesms/messages/TimingMessageContentProcessor.kt

@@ -0,0 +1,25 @@
+package org.thoughtcrime.securesms.messages
+
+import android.content.Context
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.testing.LogPredicate
+import org.whispersystems.signalservice.api.messages.SignalServiceContent
+
+class TimingMessageContentProcessor(context: Context) : MessageContentProcessor(context) {
+  companion object {
+    val TAG = Log.tag(TimingMessageContentProcessor::class.java)
+
+    fun endTagPredicate(timestamp: Long): LogPredicate = { entry ->
+      entry.tag == TAG && entry.message == endTag(timestamp)
+    }
+
+    private fun startTag(timestamp: Long) = "$timestamp start"
+    fun endTag(timestamp: Long) = "$timestamp end"
+  }
+
+  override fun process(messageState: MessageState?, content: SignalServiceContent?, exceptionMetadata: ExceptionMetadata?, envelopeTimestamp: Long, smsMessageId: Long) {
+    Log.d(TAG, startTag(envelopeTimestamp))
+    super.process(messageState, content, exceptionMetadata, envelopeTimestamp, smsMessageId)
+    Log.d(TAG, endTag(envelopeTimestamp))
+  }
+}

app/src/androidTest/java/org/thoughtcrime/securesms/testing/AliceClient.kt

@@ -0,0 +1,44 @@
+package org.thoughtcrime.securesms.testing
+
+import org.signal.libsignal.protocol.ecc.ECKeyPair
+import org.signal.libsignal.zkgroup.profiles.ProfileKey
+import org.thoughtcrime.securesms.crypto.ProfileKeyUtil
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.testing.FakeClientHelpers.toSignalServiceEnvelope
+import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.api.push.SignalServiceAddress
+
+/**
+ * Welcome to Alice's Client.
+ *
+ * Alice represent the Android instrumentation test user. Unlike [BobClient] much less is needed here
+ * as it can make use of the standard Signal Android App infrastructure.
+ */
+class AliceClient(val serviceId: ServiceId, val e164: String, val trustRoot: ECKeyPair) {
+
+  private val aliceSenderCertificate = FakeClientHelpers.createCertificateFor(
+    trustRoot = trustRoot,
+    uuid = serviceId.uuid(),
+    e164 = e164,
+    deviceId = 1,
+    identityKey = SignalStore.account().aciIdentityKey.publicKey.publicKey,
+    expires = 31337
+  )
+
+  fun process(envelope: SignalServiceEnvelope) {
+    ApplicationDependencies.getIncomingMessageProcessor().acquire().use { processor -> processor.processEnvelope(envelope) }
+  }
+
+  fun encrypt(now: Long, destination: Recipient): SignalServiceEnvelope {
+    return ApplicationDependencies.getSignalServiceMessageSender().getEncryptedMessage(
+      SignalServiceAddress(destination.requireServiceId(), destination.requireE164()),
+      FakeClientHelpers.getTargetUnidentifiedAccess(ProfileKeyUtil.getSelfProfileKey(), ProfileKey(destination.profileKey), aliceSenderCertificate),
+      1,
+      FakeClientHelpers.encryptedTextMessage(now),
+      false
+    ).toSignalServiceEnvelope(now, destination.requireServiceId())
+  }
+}

app/src/androidTest/java/org/thoughtcrime/securesms/testing/BobClient.kt

@@ -0,0 +1,167 @@
+package org.thoughtcrime.securesms.testing
+
+import org.signal.core.util.readToSingleInt
+import org.signal.core.util.select
+import org.signal.libsignal.protocol.IdentityKey
+import org.signal.libsignal.protocol.IdentityKeyPair
+import org.signal.libsignal.protocol.SessionBuilder
+import org.signal.libsignal.protocol.SignalProtocolAddress
+import org.signal.libsignal.protocol.ecc.ECKeyPair
+import org.signal.libsignal.protocol.groups.state.SenderKeyRecord
+import org.signal.libsignal.protocol.state.IdentityKeyStore
+import org.signal.libsignal.protocol.state.PreKeyBundle
+import org.signal.libsignal.protocol.state.PreKeyRecord
+import org.signal.libsignal.protocol.state.SessionRecord
+import org.signal.libsignal.protocol.state.SignedPreKeyRecord
+import org.signal.libsignal.protocol.util.KeyHelper
+import org.signal.libsignal.zkgroup.profiles.ProfileKey
+import org.thoughtcrime.securesms.crypto.ProfileKeyUtil
+import org.thoughtcrime.securesms.crypto.UnidentifiedAccessUtil
+import org.thoughtcrime.securesms.database.OneTimePreKeyTable
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.SignedPreKeyTable
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.testing.FakeClientHelpers.toSignalServiceEnvelope
+import org.whispersystems.signalservice.api.SignalServiceAccountDataStore
+import org.whispersystems.signalservice.api.SignalSessionLock
+import org.whispersystems.signalservice.api.crypto.SignalServiceCipher
+import org.whispersystems.signalservice.api.crypto.SignalSessionBuilder
+import org.whispersystems.signalservice.api.crypto.UnidentifiedAccess
+import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
+import org.whispersystems.signalservice.api.push.DistributionId
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.api.push.SignalServiceAddress
+import java.util.Optional
+import java.util.UUID
+import java.util.concurrent.locks.ReentrantLock
+
+/**
+ * Welcome to Bob's Client.
+ *
+ * Bob is a "fake" client that can start a session with the Android instrumentation test user (Alice).
+ *
+ * Bob can create a new session using a prekey bundle created from Alice's prekeys, send a message, decrypt
+ * a return message from Alice, and that'll start a standard Signal session with normal keys/ratcheting.
+ */
+class BobClient(val serviceId: ServiceId, val e164: String, val identityKeyPair: IdentityKeyPair, val trustRoot: ECKeyPair, val profileKey: ProfileKey) {
+
+  private val serviceAddress = SignalServiceAddress(serviceId, e164)
+  private val registrationId = KeyHelper.generateRegistrationId(false)
+  private val aciStore = BobSignalServiceAccountDataStore(registrationId, identityKeyPair)
+  private val senderCertificate = FakeClientHelpers.createCertificateFor(trustRoot, serviceId.uuid(), e164, 1, identityKeyPair.publicKey.publicKey, 31337)
+  private val sessionLock = object : SignalSessionLock {
+    private val lock = ReentrantLock()
+
+    override fun acquire(): SignalSessionLock.Lock {
+      lock.lock()
+      return SignalSessionLock.Lock { lock.unlock() }
+    }
+  }
+
+  /** Inspired by SignalServiceMessageSender#getEncryptedMessage */
+  fun encrypt(now: Long): SignalServiceEnvelope {
+    val envelopeContent = FakeClientHelpers.encryptedTextMessage(now)
+
+    val cipher = SignalServiceCipher(serviceAddress, 1, aciStore, sessionLock, null)
+
+    if (!aciStore.containsSession(getAliceProtocolAddress())) {
+      val sessionBuilder = SignalSessionBuilder(sessionLock, SessionBuilder(aciStore, getAliceProtocolAddress()))
+      sessionBuilder.process(getAlicePreKeyBundle())
+    }
+
+    return cipher.encrypt(getAliceProtocolAddress(), getAliceUnidentifiedAccess(), envelopeContent)
+      .toSignalServiceEnvelope(envelopeContent.content.get().dataMessage.timestamp, getAliceServiceId())
+  }
+
+  fun decrypt(envelope: SignalServiceEnvelope) {
+    val cipher = SignalServiceCipher(serviceAddress, 1, aciStore, sessionLock, UnidentifiedAccessUtil.getCertificateValidator())
+    cipher.decrypt(envelope)
+  }
+
+  private fun getAliceServiceId(): ServiceId {
+    return SignalStore.account().requireAci()
+  }
+
+  private fun getAlicePreKeyBundle(): PreKeyBundle {
+    val selfPreKeyId = SignalDatabase.rawDatabase
+      .select(OneTimePreKeyTable.KEY_ID)
+      .from(OneTimePreKeyTable.TABLE_NAME)
+      .where("${OneTimePreKeyTable.ACCOUNT_ID} = ?", getAliceServiceId().toString())
+      .run()
+      .readToSingleInt(-1)
+
+    val selfPreKeyRecord = SignalDatabase.oneTimePreKeys.get(getAliceServiceId(), selfPreKeyId)!!
+
+    val selfSignedPreKeyId = SignalDatabase.rawDatabase
+      .select(SignedPreKeyTable.KEY_ID)
+      .from(SignedPreKeyTable.TABLE_NAME)
+      .where("${SignedPreKeyTable.ACCOUNT_ID} = ?", getAliceServiceId().toString())
+      .run()
+      .readToSingleInt(-1)
+
+    val selfSignedPreKeyRecord = SignalDatabase.signedPreKeys.get(getAliceServiceId(), selfSignedPreKeyId)!!
+
+    return PreKeyBundle(
+      SignalStore.account().registrationId,
+      1,
+      selfPreKeyId,
+      selfPreKeyRecord.keyPair.publicKey,
+      selfSignedPreKeyId,
+      selfSignedPreKeyRecord.keyPair.publicKey,
+      selfSignedPreKeyRecord.signature,
+      getAlicePublicKey()
+    )
+  }
+
+  private fun getAliceProtocolAddress(): SignalProtocolAddress {
+    return SignalProtocolAddress(SignalStore.account().requireAci().toString(), 1)
+  }
+
+  private fun getAlicePublicKey(): IdentityKey {
+    return SignalStore.account().aciIdentityKey.publicKey
+  }
+
+  private fun getAliceProfileKey(): ProfileKey {
+    return ProfileKeyUtil.getSelfProfileKey()
+  }
+
+  private fun getAliceUnidentifiedAccess(): Optional<UnidentifiedAccess> {
+    return FakeClientHelpers.getTargetUnidentifiedAccess(profileKey, getAliceProfileKey(), senderCertificate)
+  }
+
+  private class BobSignalServiceAccountDataStore(private val registrationId: Int, private val identityKeyPair: IdentityKeyPair) : SignalServiceAccountDataStore {
+    private var aliceSessionRecord: SessionRecord? = null
+
+    override fun getIdentityKeyPair(): IdentityKeyPair = identityKeyPair
+
+    override fun getLocalRegistrationId(): Int = registrationId
+    override fun isTrustedIdentity(address: SignalProtocolAddress?, identityKey: IdentityKey?, direction: IdentityKeyStore.Direction?): Boolean = true
+    override fun loadSession(address: SignalProtocolAddress?): SessionRecord = aliceSessionRecord ?: SessionRecord()
+    override fun saveIdentity(address: SignalProtocolAddress?, identityKey: IdentityKey?): Boolean = false
+    override fun storeSession(address: SignalProtocolAddress?, record: SessionRecord?) { aliceSessionRecord = record }
+    override fun getSubDeviceSessions(name: String?): List<Int> = emptyList()
+    override fun containsSession(address: SignalProtocolAddress?): Boolean = aliceSessionRecord != null
+    override fun getIdentity(address: SignalProtocolAddress?): IdentityKey = SignalStore.account().aciIdentityKey.publicKey
+
+    override fun loadPreKey(preKeyId: Int): PreKeyRecord = throw UnsupportedOperationException()
+    override fun storePreKey(preKeyId: Int, record: PreKeyRecord?) = throw UnsupportedOperationException()
+    override fun containsPreKey(preKeyId: Int): Boolean = throw UnsupportedOperationException()
+    override fun removePreKey(preKeyId: Int) = throw UnsupportedOperationException()
+    override fun loadExistingSessions(addresses: MutableList<SignalProtocolAddress>?): MutableList<SessionRecord> = throw UnsupportedOperationException()
+    override fun deleteSession(address: SignalProtocolAddress?) = throw UnsupportedOperationException()
+    override fun deleteAllSessions(name: String?) = throw UnsupportedOperationException()
+    override fun loadSignedPreKey(signedPreKeyId: Int): SignedPreKeyRecord = throw UnsupportedOperationException()
+    override fun loadSignedPreKeys(): MutableList<SignedPreKeyRecord> = throw UnsupportedOperationException()
+    override fun storeSignedPreKey(signedPreKeyId: Int, record: SignedPreKeyRecord?) = throw UnsupportedOperationException()
+    override fun containsSignedPreKey(signedPreKeyId: Int): Boolean = throw UnsupportedOperationException()
+    override fun removeSignedPreKey(signedPreKeyId: Int) = throw UnsupportedOperationException()
+    override fun storeSenderKey(sender: SignalProtocolAddress?, distributionId: UUID?, record: SenderKeyRecord?) = throw UnsupportedOperationException()
+    override fun loadSenderKey(sender: SignalProtocolAddress?, distributionId: UUID?): SenderKeyRecord = throw UnsupportedOperationException()
+    override fun archiveSession(address: SignalProtocolAddress?) = throw UnsupportedOperationException()
+    override fun getAllAddressesWithActiveSessions(addressNames: MutableList<String>?): MutableSet<SignalProtocolAddress> = throw UnsupportedOperationException()
+    override fun getSenderKeySharedWith(distributionId: DistributionId?): MutableSet<SignalProtocolAddress> = throw UnsupportedOperationException()
+    override fun markSenderKeySharedWith(distributionId: DistributionId?, addresses: MutableCollection<SignalProtocolAddress>?) = throw UnsupportedOperationException()
+    override fun clearSenderKeySharedWith(addresses: MutableCollection<SignalProtocolAddress>?) = throw UnsupportedOperationException()
+    override fun isMultiDevice(): Boolean = throw UnsupportedOperationException()
+  }
+}

app/src/androidTest/java/org/thoughtcrime/securesms/testing/FakeClientHelpers.kt

@@ -0,0 +1,81 @@
+package org.thoughtcrime.securesms.testing
+
+import org.signal.libsignal.internal.Native
+import org.signal.libsignal.internal.NativeHandleGuard
+import org.signal.libsignal.metadata.certificate.CertificateValidator
+import org.signal.libsignal.metadata.certificate.SenderCertificate
+import org.signal.libsignal.metadata.certificate.ServerCertificate
+import org.signal.libsignal.protocol.ecc.Curve
+import org.signal.libsignal.protocol.ecc.ECKeyPair
+import org.signal.libsignal.protocol.ecc.ECPublicKey
+import org.signal.libsignal.zkgroup.profiles.ProfileKey
+import org.whispersystems.signalservice.api.crypto.ContentHint
+import org.whispersystems.signalservice.api.crypto.EnvelopeContent
+import org.whispersystems.signalservice.api.crypto.UnidentifiedAccess
+import org.whispersystems.signalservice.api.crypto.UnidentifiedAccessPair
+import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.internal.push.OutgoingPushMessage
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.util.Base64
+import java.util.Optional
+import java.util.UUID
+
+object FakeClientHelpers {
+
+  val noOpCertificateValidator = object : CertificateValidator(null) {
+    override fun validate(certificate: SenderCertificate, validationTime: Long) = Unit
+  }
+
+  fun createCertificateFor(trustRoot: ECKeyPair, uuid: UUID, e164: String, deviceId: Int, identityKey: ECPublicKey, expires: Long): SenderCertificate {
+    val serverKey: ECKeyPair = Curve.generateKeyPair()
+    NativeHandleGuard(serverKey.publicKey).use { serverPublicGuard ->
+      NativeHandleGuard(trustRoot.privateKey).use { trustRootPrivateGuard ->
+        val serverCertificate = ServerCertificate(Native.ServerCertificate_New(1, serverPublicGuard.nativeHandle(), trustRootPrivateGuard.nativeHandle()))
+        NativeHandleGuard(identityKey).use { identityGuard ->
+          NativeHandleGuard(serverCertificate).use { serverCertificateGuard ->
+            NativeHandleGuard(serverKey.privateKey).use { serverPrivateGuard ->
+              return SenderCertificate(Native.SenderCertificate_New(uuid.toString(), e164, deviceId, identityGuard.nativeHandle(), expires, serverCertificateGuard.nativeHandle(), serverPrivateGuard.nativeHandle()))
+            }
+          }
+        }
+      }
+    }
+  }
+
+  fun getTargetUnidentifiedAccess(myProfileKey: ProfileKey, theirProfileKey: ProfileKey, senderCertificate: SenderCertificate): Optional<UnidentifiedAccess> {
+    val selfUnidentifiedAccessKey = UnidentifiedAccess.deriveAccessKeyFrom(myProfileKey)
+    val themUnidentifiedAccessKey = UnidentifiedAccess.deriveAccessKeyFrom(theirProfileKey)
+
+    return UnidentifiedAccessPair(UnidentifiedAccess(selfUnidentifiedAccessKey, senderCertificate.serialized), UnidentifiedAccess(themUnidentifiedAccessKey, senderCertificate.serialized)).targetUnidentifiedAccess
+  }
+
+  fun encryptedTextMessage(now: Long, message: String = "Test body message"): EnvelopeContent {
+    val content = SignalServiceProtos.Content.newBuilder().apply {
+      setDataMessage(
+        SignalServiceProtos.DataMessage.newBuilder().apply {
+          body = message
+          timestamp = now
+        }
+      )
+    }
+    return EnvelopeContent.encrypted(content.build(), ContentHint.RESENDABLE, Optional.empty())
+  }
+
+  fun OutgoingPushMessage.toSignalServiceEnvelope(timestamp: Long, destination: ServiceId): SignalServiceEnvelope {
+    return SignalServiceEnvelope(
+      this.type,
+      Optional.empty(),
+      1,
+      timestamp,
+      Base64.decode(this.content),
+      timestamp + 1,
+      timestamp + 2,
+      UUID.randomUUID().toString(),
+      destination.toString(),
+      true,
+      false,
+      null
+    )
+  }
+}

app/src/androidTest/java/org/thoughtcrime/securesms/testing/InMemoryLogger.kt

@@ -0,0 +1,86 @@
+package org.thoughtcrime.securesms.testing
+
+import org.signal.core.util.concurrent.SignalExecutors
+import org.signal.core.util.logging.Log
+import java.util.concurrent.CountDownLatch
+
+typealias LogPredicate = (Entry) -> Boolean
+
+/**
+ * Logging implementation that holds logs in memory as they are added to be retrieve at a later time by a test.
+ * Can also be used for multithreaded synchronization and waiting until certain logs are emitted before continuing
+ * a test.
+ */
+class InMemoryLogger : Log.Logger() {
+
+  private val executor = SignalExecutors.newCachedSingleThreadExecutor("inmemory-logger")
+  private val predicates = mutableListOf<LogPredicate>()
+  private val logEntries = mutableListOf<Entry>()
+
+  override fun v(tag: String, message: String?, t: Throwable?, keepLonger: Boolean) = add(Verbose(tag, message, t, System.currentTimeMillis()))
+  override fun d(tag: String, message: String?, t: Throwable?, keepLonger: Boolean) = add(Debug(tag, message, t, System.currentTimeMillis()))
+  override fun i(tag: String, message: String?, t: Throwable?, keepLonger: Boolean) = add(Info(tag, message, t, System.currentTimeMillis()))
+  override fun w(tag: String, message: String?, t: Throwable?, keepLonger: Boolean) = add(Warn(tag, message, t, System.currentTimeMillis()))
+  override fun e(tag: String, message: String?, t: Throwable?, keepLonger: Boolean) = add(Error(tag, message, t, System.currentTimeMillis()))
+
+  override fun flush() {
+    val latch = CountDownLatch(1)
+    executor.execute { latch.countDown() }
+    latch.await()
+  }
+
+  private fun add(entry: Entry) {
+    executor.execute {
+      logEntries += entry
+
+      val iterator = predicates.iterator()
+      while (iterator.hasNext()) {
+        val predicate = iterator.next()
+        if (predicate(entry)) {
+          iterator.remove()
+        }
+      }
+    }
+  }
+
+  /** Blocks until a snapshot of all log entries can be taken in a thread-safe way. */
+  fun entries(): List<Entry> {
+    val latch = CountDownLatch(1)
+    var entries: List<Entry> = emptyList()
+    executor.execute {
+      entries = logEntries.toList()
+      latch.countDown()
+    }
+    latch.await()
+    return entries
+  }
+
+  /** Returns a countdown latch that'll fire at a future point when an [Entry] is received that matches the predicate. */
+  fun getLockForUntil(predicate: LogPredicate): CountDownLatch {
+    val latch = CountDownLatch(1)
+    executor.execute {
+      predicates += { entry ->
+        if (predicate(entry)) {
+          latch.countDown()
+          true
+        } else {
+          false
+        }
+      }
+    }
+    return latch
+  }
+}
+
+sealed interface Entry {
+  val tag: String
+  val message: String?
+  val throwable: Throwable?
+  val timestamp: Long
+}
+
+data class Verbose(override val tag: String, override val message: String?, override val throwable: Throwable?, override val timestamp: Long) : Entry
+data class Debug(override val tag: String, override val message: String?, override val throwable: Throwable?, override val timestamp: Long) : Entry
+data class Info(override val tag: String, override val message: String?, override val throwable: Throwable?, override val timestamp: Long) : Entry
+data class Warn(override val tag: String, override val message: String?, override val throwable: Throwable?, override val timestamp: Long) : Entry
+data class Error(override val tag: String, override val message: String?, override val throwable: Throwable?, override val timestamp: Long) : Entry

app/src/androidTest/java/org/thoughtcrime/securesms/testing/SignalActivityRule.kt

@@ -11,7 +11,9 @@ import androidx.test.platform.app.InstrumentationRegistry
 import okhttp3.mockwebserver.MockResponse
 import org.junit.rules.ExternalResource
 import org.signal.libsignal.protocol.IdentityKey
+import org.signal.libsignal.protocol.IdentityKeyPair
 import org.signal.libsignal.protocol.SignalProtocolAddress
+import org.thoughtcrime.securesms.SignalInstrumentationApplicationContext
 import org.thoughtcrime.securesms.crypto.IdentityKeyUtil
 import org.thoughtcrime.securesms.crypto.MasterSecretUtil
 import org.thoughtcrime.securesms.crypto.ProfileKeyUtil
@@ -54,11 +56,18 @@ class SignalActivityRule(private val othersCount: Int = 4) : ExternalResource()
     private set
   lateinit var others: List<RecipientId>
     private set
+  lateinit var othersKeys: List<IdentityKeyPair>
+
+  val inMemoryLogger: InMemoryLogger
+    get() = (application as SignalInstrumentationApplicationContext).inMemoryLogger
 
   override fun before() {
     context = InstrumentationRegistry.getInstrumentation().targetContext
     self = setupSelf()
-    others = setupOthers()
+
+    val setupOthers = setupOthers()
+    others = setupOthers.first
+    othersKeys = setupOthers.second
 
     InstrumentationApplicationDependencyProvider.clearHandlers()
   }
@@ -99,8 +108,9 @@ class SignalActivityRule(private val othersCount: Int = 4) : ExternalResource()
     return Recipient.self()
   }
 
-  private fun setupOthers(): List<RecipientId> {
+  private fun setupOthers(): Pair<List<RecipientId>, List<IdentityKeyPair>> {
     val others = mutableListOf<RecipientId>()
+    val othersKeys = mutableListOf<IdentityKeyPair>()
 
     if (othersCount !in 0 until 1000) {
       throw IllegalArgumentException("$othersCount must be between 0 and 1000")
@@ -114,11 +124,13 @@ class SignalActivityRule(private val othersCount: Int = 4) : ExternalResource()
       SignalDatabase.recipients.setCapabilities(recipientId, SignalServiceProfile.Capabilities(true, true, true, true, true, true, true, true, true))
       SignalDatabase.recipients.setProfileSharing(recipientId, true)
       SignalDatabase.recipients.markRegistered(recipientId, aci)
-      ApplicationDependencies.getProtocolStore().aci().saveIdentity(SignalProtocolAddress(aci.toString(), 0), IdentityKeyUtil.generateIdentityKeyPair().publicKey)
+      val otherIdentity = IdentityKeyUtil.generateIdentityKeyPair()
+      ApplicationDependencies.getProtocolStore().aci().saveIdentity(SignalProtocolAddress(aci.toString(), 0), otherIdentity.publicKey)
       others += recipientId
+      othersKeys += otherIdentity
     }
 
-    return others
+    return others to othersKeys
   }
 
   inline fun <reified T : Activity> launchActivity(initIntent: Intent.() -> Unit = {}): ActivityScenario<T> {

app/src/androidTest/java/org/thoughtcrime/securesms/testing/TestUtils.kt

@@ -7,6 +7,9 @@ import org.hamcrest.Matchers.not
 import org.hamcrest.Matchers.notNullValue
 import org.hamcrest.Matchers.nullValue
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+import kotlin.time.Duration
 
 /**
  * Run the given [runnable] on a new thread and wait for it to finish.
@@ -44,3 +47,9 @@ infix fun <T : Any> T.assertIsNot(expected: T) {
 infix fun <E, T : Collection<E>> T.assertIsSize(expected: Int) {
   assertThat(this, hasSize(expected))
 }
+
+fun CountDownLatch.awaitFor(duration: Duration) {
+  if (!await(duration.inWholeMilliseconds, TimeUnit.MILLISECONDS)) {
+    throw TimeoutException("Latch await took longer than ${duration.inWholeMilliseconds}ms")
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/ApplicationContext.java

@@ -126,7 +126,8 @@ public class ApplicationContext extends MultiDexApplication implements AppForegr
 
   private static final String TAG = Log.tag(ApplicationContext.class);
 
-  private PersistentLogger persistentLogger;
+  @VisibleForTesting
+  protected PersistentLogger persistentLogger;
 
   public static ApplicationContext getInstance(Context context) {
     return (ApplicationContext)context.getApplicationContext();
@@ -299,7 +300,8 @@ public class ApplicationContext extends MultiDexApplication implements AppForegr
     }
   }
 
-  private void initializeLogging() {
+  @VisibleForTesting
+  protected void initializeLogging() {
     persistentLogger = new PersistentLogger(this);
     org.signal.core.util.logging.Log.initialize(FeatureFlags::internalUser, new AndroidLogger(), persistentLogger);
 

app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.java

@@ -8,6 +8,7 @@ import android.text.TextUtils;
 
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
+import androidx.annotation.VisibleForTesting;
 
 import com.annimon.stream.Collectors;
 import com.annimon.stream.Stream;
@@ -198,7 +199,7 @@ import java.util.concurrent.TimeUnit;
  * data to our data stores.
  */
 @SuppressWarnings({ "OptionalGetWithoutIsPresent", "OptionalIsPresent" })
-public final class MessageContentProcessor {
+public class MessageContentProcessor {
 
   private static final String TAG = Log.tag(MessageContentProcessor.class);
 
@@ -208,7 +209,8 @@ public final class MessageContentProcessor {
     return new MessageContentProcessor(context);
   }
 
-  private MessageContentProcessor(@NonNull Context context) {
+  @VisibleForTesting
+  MessageContentProcessor(@NonNull Context context) {
     this.context = context;
   }
 

app/src/main/java/org/thoughtcrime/securesms/registration/RegistrationUtil.java

@@ -1,9 +1,5 @@
 package org.thoughtcrime.securesms.registration;
 
-import android.content.Context;
-
-import androidx.annotation.NonNull;
-
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.jobs.DirectoryRefreshJob;

dependencies.gradle

@@ -169,6 +169,7 @@ dependencyResolutionManagement {
             alias('assertj-core').to('org.assertj:assertj-core:3.11.1')
             alias('square-okhttp-mockserver').to('com.squareup.okhttp3:mockwebserver:3.12.13')
             alias('mockk').to('io.mockk:mockk:1.13.2')
+            alias('mockk-android').to('io.mockk:mockk-android:1.13.2')
 
             alias('conscrypt-openjdk-uber').to('org.conscrypt:conscrypt-openjdk-uber:2.0.0')
         }

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

@@ -2209,7 +2209,8 @@ public class SignalServiceMessageSender {
     return new OutgoingPushMessageList(recipient.getIdentifier(), timestamp, messages, online, urgent);
   }
 
-  private OutgoingPushMessage getEncryptedMessage(SignalServiceAddress         recipient,
+  // Visible for testing only
+  public OutgoingPushMessage getEncryptedMessage(SignalServiceAddress         recipient,
                                                   Optional<UnidentifiedAccess> unidentifiedAccess,
                                                   int                          deviceId,
                                                   EnvelopeContent              plaintext,
@@ -2248,7 +2249,6 @@ public class SignalServiceMessageSender {
     }
   }
 
-
   private List<PreKeyBundle> getPreKeys(SignalServiceAddress recipient, Optional<UnidentifiedAccess> unidentifiedAccess, int deviceId, boolean story) throws IOException {
     try {
       return socket.getPreKeys(recipient, unidentifiedAccess, deviceId);

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

@@ -12,13 +12,13 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 public class OutgoingPushMessage {
 
   @JsonProperty
-  private int    type;
+  public int    type;
   @JsonProperty
-  private int    destinationDeviceId;
+  public int    destinationDeviceId;
   @JsonProperty
-  private int    destinationRegistrationId;
+  public int    destinationRegistrationId;
   @JsonProperty
-  private String content;
+  public String content;
 
   public OutgoingPushMessage() {}