ec66f3b21e
The purpose of this change is to alleviate the main source of CPU-grinding after sending a message to a large group: about 20% of the total time is spent decoding and re-encoding messages as they are fetched from the DB and then updated with info about which recipients have acknowledged delivery. Prior to this commit, we decrypted the payload and then processed the message in a single rather deep call tree that began in `processNextBatch`. There are two completely different paths through this tree that lead to handling a delivery receipt because server-generated delivery receipts have the necessary info in plaintext in the envelope and client-generated delivery receipts have it in the ciphertext. The idea here is to break incoming envelope handling into two parts. In the first part, each envelope is decrypted and other common activities (like handling sender key distribution messages) are performed. A processing request is created for each. Next, processing requests are handled. It's easy to examine a processing request to determine if it is a delivery receipt, and if so, for what outgoing message. Doing so allows the new DeliveryReceiptContext class to cache message fetches and combine updates to the same message. Processing requests are grouped together so that sequential delivery receipt requests enjoy caching of messages and coalescing of updates into a single fetch/decode/encode/commit. Other kinds of envelopes are handled immediately to avoid increasing memory pressure that caching multiple messages could cause.
389 lines
19 KiB
Swift
389 lines
19 KiB
Swift
//
|
|
// Copyright (c) 2022 Open Whisper Systems. All rights reserved.
|
|
//
|
|
|
|
import XCTest
|
|
@testable import SignalServiceKit
|
|
import GRDB
|
|
import LibSignalClient
|
|
|
|
class MessageProcessingIntegrationTest: SSKBaseTestSwift {
|
|
|
|
let localE164Identifier = "+13235551234"
|
|
let localUUID = UUID()
|
|
|
|
let aliceE164Identifier = "+14715355555"
|
|
var aliceClient: TestSignalClient!
|
|
|
|
let bobE164Identifier = "+18083235555"
|
|
var bobClient: TestSignalClient!
|
|
|
|
let localClient = LocalSignalClient()
|
|
let runner = TestProtocolRunner()
|
|
lazy var fakeService = FakeService(localClient: localClient, runner: runner)
|
|
|
|
// MARK: - Hooks
|
|
|
|
override func setUp() {
|
|
super.setUp()
|
|
|
|
// Use DatabaseChangeObserver to be notified of DB writes so we
|
|
// can verify the expected changes occur.
|
|
try! databaseStorage.grdbStorage.setupDatabaseChangeObserver()
|
|
|
|
// ensure local client has necessary "registered" state
|
|
identityManager.generateNewIdentityKey(for: .aci)
|
|
identityManager.generateNewIdentityKey(for: .pni)
|
|
tsAccountManager.registerForTests(withLocalNumber: localE164Identifier, uuid: localUUID, pni: UUID())
|
|
|
|
bobClient = FakeSignalClient.generate(e164Identifier: bobE164Identifier)
|
|
aliceClient = FakeSignalClient.generate(e164Identifier: aliceE164Identifier)
|
|
}
|
|
|
|
override func tearDown() {
|
|
databaseStorage.grdbStorage.testing_tearDownDatabaseChangeObserver()
|
|
|
|
super.tearDown()
|
|
}
|
|
|
|
// MARK: - Tests
|
|
|
|
func test_contactMessage_e164AndUuidEnvelope() {
|
|
write { transaction in
|
|
try! self.runner.initialize(senderClient: self.bobClient,
|
|
recipientClient: self.localClient,
|
|
transaction: transaction)
|
|
}
|
|
|
|
// Wait until message processing has completed, otherwise future
|
|
// tests may break as we try and drain the processing queue.
|
|
let expectFlushNotification = expectation(description: "queue flushed")
|
|
NotificationCenter.default.observe(once: MessageProcessor.messageProcessorDidFlushQueue).done { _ in
|
|
expectFlushNotification.fulfill()
|
|
}
|
|
|
|
let expectMessageProcessed = expectation(description: "message processed")
|
|
// This test fulfills an expectation when a write to the database causes the desired state to be reached.
|
|
// However, there may still be writes to the database in flight, and the *next* write will also probably
|
|
// be in the desired state, resulting in the expectation being fulfilled again.
|
|
expectMessageProcessed.assertForOverFulfill = false
|
|
|
|
read { transaction in
|
|
XCTAssertEqual(0, TSMessage.anyCount(transaction: transaction))
|
|
XCTAssertEqual(0, TSThread.anyCount(transaction: transaction))
|
|
}
|
|
|
|
let databaseDelegate = DatabaseWriteBlockDelegate { _ in
|
|
self.read { transaction in
|
|
// Each time a write occurs, check to see if we've achieved the expected DB state.
|
|
//
|
|
// There are multiple writes that occur before the desired state is achieved, but
|
|
// this block is called after each one, so it must be forgiving for the prior writes.
|
|
if let message = TSMessage.anyFetchAll(transaction: transaction).first as? TSIncomingMessage {
|
|
XCTAssertEqual(1, TSMessage.anyCount(transaction: transaction))
|
|
XCTAssertEqual(message.authorAddress, self.bobClient.address)
|
|
XCTAssertNotEqual(message.authorAddress, self.aliceClient.address)
|
|
XCTAssertEqual(message.body, "Those who stands for nothing will fall for anything")
|
|
XCTAssertEqual(1, TSThread.anyCount(transaction: transaction))
|
|
guard let thread = TSThread.anyFetchAll(transaction: transaction).first as? TSContactThread else {
|
|
XCTFail("thread was unexpectedly nil")
|
|
return
|
|
}
|
|
XCTAssertEqual(thread.contactAddress, self.bobClient.address)
|
|
XCTAssertNotEqual(thread.contactAddress, self.aliceClient.address)
|
|
expectMessageProcessed.fulfill()
|
|
}
|
|
}
|
|
}
|
|
guard let observer = databaseStorage.grdbStorage.databaseChangeObserver else {
|
|
owsFailDebug("observer was unexpectedly nil")
|
|
return
|
|
}
|
|
observer.appendDatabaseWriteDelegate(databaseDelegate)
|
|
|
|
let envelopeBuilder = try! fakeService.envelopeBuilder(fromSenderClient: bobClient, bodyText: "Those who stands for nothing will fall for anything")
|
|
envelopeBuilder.setSourceE164(bobClient.e164Identifier!)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
let envelopeData = try! envelopeBuilder.buildSerializedData()
|
|
messageProcessor.processEncryptedEnvelopeData(envelopeData,
|
|
serverDeliveryTimestamp: NSDate.ows_millisecondTimeStamp(),
|
|
envelopeSource: .tests) { error in
|
|
switch error {
|
|
case MessageProcessingError.duplicatePendingEnvelope?:
|
|
XCTFail("duplicatePendingEnvelope")
|
|
case .some:
|
|
XCTFail("failure")
|
|
case nil:
|
|
break
|
|
}
|
|
}
|
|
|
|
waitForExpectations(timeout: 1.0)
|
|
}
|
|
|
|
func test_contactMessage_UuidOnlyEnvelope() {
|
|
write { transaction in
|
|
try! self.runner.initialize(senderClient: self.bobClient,
|
|
recipientClient: self.localClient,
|
|
transaction: transaction)
|
|
}
|
|
|
|
// Wait until message processing has completed, otherwise future
|
|
// tests may break as we try and drain the processing queue.
|
|
let expectFlushNotification = expectation(description: "queue flushed")
|
|
NotificationCenter.default.observe(once: MessageProcessor.messageProcessorDidFlushQueue).done { _ in
|
|
expectFlushNotification.fulfill()
|
|
}
|
|
|
|
let expectMessageProcessed = expectation(description: "message processed")
|
|
// This test fulfills an expectation when a write to the database causes the desired state to be reached.
|
|
// However, there may still be writes to the database in flight, and the *next* write will also probably
|
|
// be in the desired state, resulting in the expectation being fulfilled again.
|
|
expectMessageProcessed.assertForOverFulfill = false
|
|
|
|
read { transaction in
|
|
XCTAssertEqual(0, TSMessage.anyCount(transaction: transaction))
|
|
XCTAssertEqual(0, TSThread.anyCount(transaction: transaction))
|
|
}
|
|
|
|
let snapshotDelegate = DatabaseWriteBlockDelegate { _ in
|
|
self.read { transaction in
|
|
// Each time a write occurs, check to see if we've achieved the expected DB state.
|
|
//
|
|
// There are multiple writes that occur before the desired state is achieved, but
|
|
// this block is called after each one, so it must be forgiving for the prior writes.
|
|
if let message = TSMessage.anyFetchAll(transaction: transaction).first as? TSIncomingMessage {
|
|
XCTAssertEqual(1, TSMessage.anyCount(transaction: transaction))
|
|
XCTAssertEqual(message.authorAddress, self.bobClient.address)
|
|
XCTAssertNotEqual(message.authorAddress, self.aliceClient.address)
|
|
XCTAssertEqual(message.body, "Those who stands for nothing will fall for anything")
|
|
XCTAssertEqual(1, TSThread.anyCount(transaction: transaction))
|
|
guard let thread = TSThread.anyFetchAll(transaction: transaction).first as? TSContactThread else {
|
|
XCTFail("thread was unexpectedly nil")
|
|
return
|
|
}
|
|
XCTAssertEqual(thread.contactAddress, self.bobClient.address)
|
|
XCTAssertNotEqual(thread.contactAddress, self.aliceClient.address)
|
|
|
|
expectMessageProcessed.fulfill()
|
|
}
|
|
}
|
|
}
|
|
guard let observer = databaseStorage.grdbStorage.databaseChangeObserver else {
|
|
owsFailDebug("observer was unexpectedly nil")
|
|
return
|
|
}
|
|
observer.appendDatabaseWriteDelegate(snapshotDelegate)
|
|
|
|
let envelopeBuilder = try! fakeService.envelopeBuilder(fromSenderClient: bobClient, bodyText: "Those who stands for nothing will fall for anything")
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
let envelopeData = try! envelopeBuilder.buildSerializedData()
|
|
messageProcessor.processEncryptedEnvelopeData(envelopeData,
|
|
serverDeliveryTimestamp: NSDate.ows_millisecondTimeStamp(),
|
|
envelopeSource: .tests) { error in
|
|
switch error {
|
|
case MessageProcessingError.duplicatePendingEnvelope?:
|
|
XCTFail("duplicatePendingEnvelope")
|
|
case .some:
|
|
XCTFail("failure")
|
|
case nil:
|
|
break
|
|
}
|
|
}
|
|
waitForExpectations(timeout: 1.0)
|
|
}
|
|
|
|
func testWrongDestinationUuid() {
|
|
write { transaction in
|
|
try! self.runner.initialize(senderClient: self.bobClient,
|
|
recipientClient: self.localClient,
|
|
transaction: transaction)
|
|
}
|
|
|
|
// Wait until message processing has completed, otherwise future
|
|
// tests may break as we try and drain the processing queue.
|
|
let expectFlushNotification = expectation(description: "queue flushed")
|
|
NotificationCenter.default.observe(once: MessageProcessor.messageProcessorDidFlushQueue).done { _ in
|
|
expectFlushNotification.fulfill()
|
|
}
|
|
|
|
let envelopeBuilder = try! fakeService.envelopeBuilder(fromSenderClient: bobClient, bodyText: "Those who stands for nothing will fall for anything")
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
envelopeBuilder.setDestinationUuid(UUID().uuidString)
|
|
let envelopeData = try! envelopeBuilder.buildSerializedData()
|
|
messageProcessor.processEncryptedEnvelopeData(envelopeData,
|
|
serverDeliveryTimestamp: NSDate.ows_millisecondTimeStamp(),
|
|
envelopeSource: .tests) { error in
|
|
switch error {
|
|
case MessageProcessingError.wrongDestinationUuid?:
|
|
break
|
|
case let error?:
|
|
XCTFail("unexpected error \(error)")
|
|
case nil:
|
|
XCTFail("should have failed")
|
|
}
|
|
}
|
|
waitForExpectations(timeout: 1.0)
|
|
}
|
|
|
|
private func deliveryReceiptProtoEnvelope(timestamp: UInt64) throws -> SSKProtoEnvelope {
|
|
let envelopeBuilder = fakeService.envelopeBuilderForServerGeneratedDeliveryReceipt(fromSenderClient: bobClient)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
|
|
envelopeBuilder.setTimestamp(timestamp)
|
|
return try envelopeBuilder.build()
|
|
}
|
|
|
|
private func invalidProtoEnvelope() throws -> SSKProtoEnvelope {
|
|
let envelopeBuilder = fakeService.envelopeBuilderForInvalidEnvelope(fromSenderClient: bobClient)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
|
|
return try envelopeBuilder.build()
|
|
}
|
|
|
|
private func udDeliveryReceiptProtoEnvelope(timestamp: UInt64) throws -> SSKProtoEnvelope {
|
|
let envelopeBuilder = fakeService.envelopeBuilderForUDDeliveryReceipt(fromSenderClient: bobClient,
|
|
timestamp: timestamp)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
|
|
return try envelopeBuilder.build()
|
|
}
|
|
|
|
private func normalMessageProtoEnvelope() throws -> SSKProtoEnvelope {
|
|
let envelopeBuilder = try! fakeService.envelopeBuilder(fromSenderClient: bobClient, bodyText: "Those who stands for nothing will fall for anything")
|
|
envelopeBuilder.setSourceE164(bobClient.e164Identifier!)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
return try envelopeBuilder.build()
|
|
}
|
|
|
|
private func decrytpedEnvelopeForDeliveryReceipt(timestamp: UInt64) throws -> DecryptedEnvelope {
|
|
let envelope = try deliveryReceiptProtoEnvelope(timestamp: timestamp)
|
|
let data = try envelope.serializedData()
|
|
return DecryptedEnvelope(envelope: envelope,
|
|
envelopeData: data,
|
|
plaintextData: nil,
|
|
serverDeliveryTimestamp: 1234,
|
|
wasReceivedByUD: false,
|
|
identity: .pni,
|
|
completion: { _ in })
|
|
}
|
|
|
|
private func decryptedEnvelopeForNormalMessage() throws -> DecryptedEnvelope {
|
|
let envelope = try normalMessageProtoEnvelope()
|
|
let data = try envelope.serializedData()
|
|
return DecryptedEnvelope(envelope: envelope,
|
|
envelopeData: data,
|
|
plaintextData: nil,
|
|
serverDeliveryTimestamp: 1234,
|
|
wasReceivedByUD: false,
|
|
identity: .pni,
|
|
completion: { _ in })
|
|
}
|
|
|
|
private func deliveryReceiptProcessingRequest(timestamp: UInt64) throws -> ProcessingRequest {
|
|
let decryptedEnvelope = try decrytpedEnvelopeForDeliveryReceipt(timestamp: 1)
|
|
return ProcessingRequest(decryptedEnvelope,
|
|
state: .plaintextReceipt(decryptedEnvelope.envelope))
|
|
}
|
|
|
|
private func normalMessageProcessingRequest() throws -> ProcessingRequest {
|
|
let envelope = try! normalMessageProtoEnvelope()
|
|
let mmr = write { transaction in
|
|
MessageManagerRequest(envelope: envelope,
|
|
plaintextData: Data(),
|
|
wasReceivedByUD: false,
|
|
serverDeliveryTimestamp: 1,
|
|
shouldDiscardVisibleMessages: false,
|
|
transaction: transaction)!
|
|
}
|
|
let decryptedEnvelope = try decrytpedEnvelopeForDeliveryReceipt(timestamp: 1)
|
|
return ProcessingRequest(decryptedEnvelope,
|
|
state: .messageManagerRequest(mmr))
|
|
}
|
|
|
|
func testPniMessage() {
|
|
let localPniClient = LocalSignalClient(identity: .pni)
|
|
write { transaction in
|
|
try! self.runner.initializePreKeys(senderClient: self.bobClient,
|
|
recipientClient: localPniClient,
|
|
transaction: transaction)
|
|
}
|
|
|
|
// Wait until message processing has completed, otherwise future
|
|
// tests may break as we try and drain the processing queue.
|
|
_ = expectation(forNotification: MessageProcessor.messageProcessorDidFlushQueue, object: nil)
|
|
|
|
read { transaction in
|
|
XCTAssertEqual(0, TSMessage.anyCount(transaction: transaction))
|
|
XCTAssertEqual(0, TSThread.anyCount(transaction: transaction))
|
|
XCTAssertFalse(self.identityManager.shouldSharePhoneNumber(with: bobClient.address,
|
|
transaction: transaction))
|
|
}
|
|
|
|
let content = try! fakeService.buildContentData(bodyText: "Those who stands for nothing will fall for anything")
|
|
let ciphertext = databaseStorage.write { transaction in
|
|
try! runner.encrypt(content,
|
|
senderClient: bobClient,
|
|
recipient: localPniClient.protocolAddress,
|
|
context: transaction)
|
|
}
|
|
|
|
let envelopeBuilder = SSKProtoEnvelope.builder(timestamp: 100)
|
|
envelopeBuilder.setContent(Data(ciphertext.serialize()))
|
|
envelopeBuilder.setType(.prekeyBundle)
|
|
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
|
|
envelopeBuilder.setSourceDevice(1)
|
|
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
|
|
envelopeBuilder.setServerGuid(UUID().uuidString)
|
|
envelopeBuilder.setDestinationUuid(tsAccountManager.localPni!.uuidString)
|
|
let envelopeData = try! envelopeBuilder.buildSerializedData()
|
|
messageProcessor.processEncryptedEnvelopeData(envelopeData,
|
|
serverDeliveryTimestamp: NSDate.ows_millisecondTimeStamp(),
|
|
envelopeSource: .tests) { error in
|
|
switch error {
|
|
case let error?:
|
|
XCTFail("failure \(error)")
|
|
case nil:
|
|
break
|
|
}
|
|
self.read { transaction in
|
|
XCTAssert(self.identityManager.shouldSharePhoneNumber(with: self.bobClient.address,
|
|
transaction: transaction))
|
|
}
|
|
}
|
|
waitForExpectations(timeout: 1.0)
|
|
}
|
|
}
|
|
|
|
// MARK: - Helpers
|
|
|
|
class DatabaseWriteBlockDelegate {
|
|
let block: (Database) -> Void
|
|
init(block: @escaping (Database) -> Void) {
|
|
self.block = block
|
|
}
|
|
}
|
|
|
|
extension DatabaseWriteBlockDelegate: DatabaseWriteDelegate {
|
|
|
|
func databaseDidChange(with event: DatabaseEvent) { /* no-op */ }
|
|
func databaseDidCommit(db: Database) {
|
|
block(db)
|
|
}
|
|
func databaseDidRollback(db: Database) { /* no-op */ }
|
|
}
|