Browse Source

Reduced unneeded DB write operations and fixed a few minor UI bugs

Updated the database to better support the application getting suspended (0xdead10cc crash)
Updated the SOGS message handling to delete messages based on a new 'deleted' flag instead of 'data' being null
Updated the code to prevent the typing indicator from needing a DB write block as frequently
Updated the code to stop any pending jobs when entering the background (in an attempt to prevent the database suspension from causing issues)
Removed the duplicate 'Capabilities.Capability' type (updated 'Capability.Variant' to work in the same way)
Fixed a bug where a number of icons (inc. the "download document" icon) were the wrong colour in dark mode
Fixed a bug where the '@You' highlight could incorrectly have it's width reduced in some cases (had protection to prevent it being larger than the line, but that is a valid case)
Fixed a bug where the JobRunner was starting the background (which could lead to trying to access the database once it had been suspended)
Updated to the latest version of GRDB
Added some logic to the BackgroundPoller process to try and stop processing if the timeout is triggered (will catch some cases but others will end up logging a bunch of "Database is suspended" errors)
Added in some protection to prevent future deferral loops in the JobRunner
pull/612/head
Morgan Pretty 2 months ago
parent
commit
1224e539ea
  1. 4
      Podfile.lock
  2. 20
      Session/Conversations/ConversationVC+Interaction.swift
  3. 23
      Session/Conversations/ConversationViewModel.swift
  4. 4
      Session/Conversations/Input View/InputViewButton.swift
  5. 2
      Session/Conversations/Message Cells/Content Views/CallMessageView.swift
  6. 4
      Session/Conversations/Message Cells/Content Views/DeletedMessageView.swift
  7. 4
      Session/Conversations/Message Cells/Content Views/MediaPlaceholderView.swift
  8. 2
      Session/Conversations/Message Cells/Content Views/OpenGroupInvitationView.swift
  9. 17
      Session/Meta/AppDelegate.swift
  10. 3
      Session/Shared/HighlightMentionBackgroundView.swift
  11. 67
      Session/Utilities/BackgroundPoller.swift
  12. 34
      SessionMessagingKit/Database/Models/Capability.swift
  13. 52
      SessionMessagingKit/Open Groups/Models/Capabilities.swift
  14. 3
      SessionMessagingKit/Open Groups/Models/SOGSMessage.swift
  15. 15
      SessionMessagingKit/Open Groups/OpenGroupManager.swift
  16. 7
      SessionMessagingKit/Sending & Receiving/Message Handling/MessageReceiver+TypingIndicators.swift
  17. 41
      SessionMessagingKit/Sending & Receiving/MessageSender.swift
  18. 80
      SessionMessagingKit/Sending & Receiving/Pollers/ClosedGroupPoller.swift
  19. 242
      SessionMessagingKit/Sending & Receiving/Pollers/OpenGroupPoller.swift
  20. 4
      SessionMessagingKit/Sending & Receiving/Pollers/Poller.swift
  21. 123
      SessionMessagingKit/Sending & Receiving/Typing Indicators/TypingIndicators.swift
  22. 37
      SessionSnodeKit/Database/Models/SnodeReceivedMessageInfo.swift
  23. 2
      SessionUtilitiesKit/Database/Storage.swift
  24. 6
      SessionUtilitiesKit/General/Dictionary+Utilities.swift
  25. 88
      SessionUtilitiesKit/JobRunner/JobRunner.swift
  26. 2
      SessionUtilitiesKit/JobRunner/JobRunnerError.swift

4
Podfile.lock

@ -27,7 +27,7 @@ PODS:
- DifferenceKit/Core (1.2.0)
- DifferenceKit/UIKitExtension (1.2.0):
- DifferenceKit/Core
- GRDB.swift/SQLCipher (5.24.1):
- GRDB.swift/SQLCipher (5.26.0):
- SQLCipher (>= 3.4.0)
- libwebp (1.2.1):
- libwebp/demux (= 1.2.1)
@ -222,7 +222,7 @@ SPEC CHECKSUMS:
CryptoSwift: a532e74ed010f8c95f611d00b8bbae42e9fe7c17
Curve25519Kit: e63f9859ede02438ae3defc5e1a87e09d1ec7ee6
DifferenceKit: 5659c430bb7fe45876fa32ce5cba5d6167f0c805
GRDB.swift: b3180ce2135fc06a453297889b746b1478c4d8c7
GRDB.swift: 1395cb3556df6b16ed69dfc74c3886abc75d2825
libwebp: 98a37e597e40bfdb4c911fc98f2c53d0b12d05fc
Nimble: 5316ef81a170ce87baf72dd961f22f89a602ff84
NVActivityIndicatorView: 1f6c5687f1171810aa27a3296814dc2d7dec3667

20
Session/Conversations/ConversationVC+Interaction.swift

@ -520,16 +520,18 @@ extension ConversationVC:
let threadId: String = self.viewModel.threadData.threadId
let threadVariant: SessionThread.Variant = self.viewModel.threadData.threadVariant
let threadIsMessageRequest: Bool = (self.viewModel.threadData.threadIsMessageRequest == true)
let needsToStartTypingIndicator: Bool = TypingIndicators.didStartTypingNeedsToStart(
threadId: threadId,
threadVariant: threadVariant,
threadIsMessageRequest: threadIsMessageRequest,
direction: .outgoing,
timestampMs: Int64(floor(Date().timeIntervalSince1970 * 1000))
)
Storage.shared.writeAsync { db in
TypingIndicators.didStartTyping(
db,
threadId: threadId,
threadVariant: threadVariant,
threadIsMessageRequest: threadIsMessageRequest,
direction: .outgoing,
timestampMs: Int64(floor(Date().timeIntervalSince1970 * 1000))
)
if needsToStartTypingIndicator {
Storage.shared.writeAsync { db in
TypingIndicators.start(db, threadId: threadId, direction: .outgoing)
}
}
}

23
Session/Conversations/ConversationViewModel.swift

@ -418,15 +418,34 @@ public class ConversationViewModel: OWSAudioPlayerDelegate {
// MARK: - Functions
public func updateDraft(to draft: String) {
let threadId: String = self.threadId
let currentDraft: String = Storage.shared
.read { db in
try SessionThread
.select(.messageDraft)
.filter(id: threadId)
.asRequest(of: String.self)
.fetchOne(db)
}
.defaulting(to: "")
// Only write the updated draft to the database if it's changed (avoid unnecessary writes)
guard draft != currentDraft else { return }
Storage.shared.writeAsync { db in
try SessionThread
.filter(id: self.threadId)
.filter(id: threadId)
.updateAll(db, SessionThread.Columns.messageDraft.set(to: draft))
}
}
public func markAllAsRead() {
guard let lastInteractionId: Int64 = self.threadData.interactionId else { return }
// Don't bother marking anything as read if there are no unread interactions (we can rely
// on the 'threadData.threadUnreadCount' to always be accurate)
guard
(self.threadData.threadUnreadCount ?? 0) > 0,
let lastInteractionId: Int64 = self.threadData.interactionId
else { return }
let threadId: String = self.threadData.threadId
let trySendReadReceipt: Bool = (self.threadData.threadIsMessageRequest == false)

4
Session/Conversations/Input View/InputViewButton.swift

@ -59,8 +59,8 @@ final class InputViewButton : UIView {
isUserInteractionEnabled = true
widthConstraint.isActive = true
heightConstraint.isActive = true
let tint = isSendButton ? UIColor.black : Colors.text
let iconImageView = UIImageView(image: icon.withTint(tint))
let iconImageView = UIImageView(image: icon.withRenderingMode(.alwaysTemplate))
iconImageView.tintColor = (isSendButton ? UIColor.black : Colors.text)
iconImageView.contentMode = .scaleAspectFit
let iconSize = InputViewButton.iconSize
iconImageView.set(.width, to: iconSize)

2
Session/Conversations/Message Cells/Content Views/CallMessageView.swift

@ -28,8 +28,8 @@ final class CallMessageView: UIView {
// Image view
let imageView: UIImageView = UIImageView(
image: UIImage(named: "Phone")?
.resizedImage(to: CGSize(width: CallMessageView.iconSize, height: CallMessageView.iconSize))?
.withRenderingMode(.alwaysTemplate)
.resizedImage(to: CGSize(width: CallMessageView.iconSize, height: CallMessageView.iconSize))
)
imageView.tintColor = textColor
imageView.contentMode = .center

4
Session/Conversations/Message Cells/Content Views/DeletedMessageView.swift

@ -27,11 +27,11 @@ final class DeletedMessageView: UIView {
private func setUpViewHierarchy(textColor: UIColor) {
// Image view
let icon = UIImage(named: "ic_trash")?
.withRenderingMode(.alwaysTemplate)
.resizedImage(to: CGSize(
width: DeletedMessageView.iconSize,
height: DeletedMessageView.iconSize
))
))?
.withRenderingMode(.alwaysTemplate)
let imageView = UIImageView(image: icon)
imageView.tintColor = textColor

4
Session/Conversations/Message Cells/Content Views/MediaPlaceholderView.swift

@ -44,13 +44,13 @@ final class MediaPlaceholderView: UIView {
// Image view
let imageView = UIImageView(
image: UIImage(named: iconName)?
.withRenderingMode(.alwaysTemplate)
.resizedImage(
to: CGSize(
width: MediaPlaceholderView.iconSize,
height: MediaPlaceholderView.iconSize
)
)
)?
.withRenderingMode(.alwaysTemplate)
)
imageView.tintColor = textColor
imageView.contentMode = .center

2
Session/Conversations/Message Cells/Content Views/OpenGroupInvitationView.swift

@ -68,8 +68,8 @@ final class OpenGroupInvitationView: UIView {
let iconImageViewSize = OpenGroupInvitationView.iconImageViewSize
let iconImageView = UIImageView(
image: UIImage(named: iconName)?
.resizedImage(to: CGSize(width: iconSize, height: iconSize))?
.withRenderingMode(.alwaysTemplate)
.resizedImage(to: CGSize(width: iconSize, height: iconSize))
)
iconImageView.tintColor = .white
iconImageView.contentMode = .center

17
Session/Meta/AppDelegate.swift

@ -122,6 +122,9 @@ class AppDelegate: UIResponder, UIApplicationDelegate, UNUserNotificationCenterD
/// `appDidFinishLaunching` seems to fix this odd behaviour (even though it doesn't match
/// Apple's documentation on the matter)
UNUserNotificationCenter.current().delegate = self
// Resume database
NotificationCenter.default.post(name: Database.resumeNotification, object: self)
}
func applicationDidEnterBackground(_ application: UIApplication) {
@ -130,6 +133,10 @@ class AppDelegate: UIResponder, UIApplicationDelegate, UNUserNotificationCenterD
// NOTE: Fix an edge case where user taps on the callkit notification
// but answers the call on another device
stopPollers(shouldStopUserPoller: !self.hasIncomingCallWaiting())
JobRunner.stopAndClearPendingJobs()
// Suspend database
NotificationCenter.default.post(name: Database.suspendNotification, object: self)
}
func applicationDidReceiveMemoryWarning(_ application: UIApplication) {
@ -185,8 +192,16 @@ class AppDelegate: UIResponder, UIApplicationDelegate, UNUserNotificationCenterD
// MARK: - Background Fetching
func application(_ application: UIApplication, performFetchWithCompletionHandler completionHandler: @escaping (UIBackgroundFetchResult) -> Void) {
// Resume database
NotificationCenter.default.post(name: Database.resumeNotification, object: self)
AppReadiness.runNowOrWhenAppDidBecomeReady {
BackgroundPoller.poll(completionHandler: completionHandler)
BackgroundPoller.poll { result in
// Suspend database
NotificationCenter.default.post(name: Database.suspendNotification, object: self)
completionHandler(result)
}
}
}

3
Session/Shared/HighlightMentionBackgroundView.swift

@ -137,9 +137,6 @@ class HighlightMentionBackgroundView: UIView {
extraYOffset
)
// We don't want to draw too far to the right
runBounds.size.width = (runBounds.width > lineWidth ? lineWidth : runBounds.width)
let path = UIBezierPath(roundedRect: runBounds, cornerRadius: cornerRadius)
mentionBackgroundColor.setFill()
path.fill()

67
Session/Utilities/BackgroundPoller.swift

@ -9,8 +9,11 @@ import SessionUtilitiesKit
public final class BackgroundPoller {
private static var promises: [Promise<Void>] = []
private static var isValid: Bool = false
public static func poll(completionHandler: @escaping (UIBackgroundFetchResult) -> Void) {
BackgroundPoller.isValid = true
promises = []
.appending(pollForMessages())
.appending(contentsOf: pollForClosedGroupMessages())
@ -32,7 +35,11 @@ public final class BackgroundPoller {
let poller: OpenGroupAPI.Poller = OpenGroupAPI.Poller(for: server)
poller.stop()
return poller.poll(isBackgroundPoll: true, isPostCapabilitiesRetry: false)
return poller.poll(
isBackgroundPoll: true,
isBackgroundPollerValid: { BackgroundPoller.isValid },
isPostCapabilitiesRetry: false
)
}
)
@ -41,6 +48,7 @@ public final class BackgroundPoller {
// after 25 seconds allowing us to cancel all pending promises
let cancelTimer: Timer = Timer.scheduledTimerOnMainThread(withTimeInterval: 25, repeats: false) { timer in
timer.invalidate()
BackgroundPoller.isValid = false
guard promises.contains(where: { !$0.isResolved }) else { return }
@ -50,6 +58,9 @@ public final class BackgroundPoller {
when(resolved: promises)
.done { _ in
// If we have already invalidated the timer then do nothing (we essentially timed out)
guard cancelTimer.isValid else { return }
cancelTimer.invalidate()
completionHandler(.newData)
}
@ -88,7 +99,8 @@ public final class BackgroundPoller {
groupPublicKey,
on: DispatchQueue.main,
maxRetryCount: 0,
isBackgroundPoll: true
isBackgroundPoll: true,
isBackgroundPollValid: { BackgroundPoller.isValid }
)
}
}
@ -100,44 +112,45 @@ public final class BackgroundPoller {
return SnodeAPI.getMessages(from: snode, associatedWith: publicKey)
.then(on: DispatchQueue.main) { messages -> Promise<Void> in
guard !messages.isEmpty else { return Promise.value(()) }
guard !messages.isEmpty, BackgroundPoller.isValid else { return Promise.value(()) }
var jobsToRun: [Job] = []
Storage.shared.write { db in
var threadMessages: [String: [MessageReceiveJob.Details.MessageInfo]] = [:]
messages.forEach { message in
do {
let processedMessage: ProcessedMessage? = try Message.processRawReceivedMessage(db, rawMessage: message)
let key: String = (processedMessage?.threadId ?? Message.nonThreadMessageId)
threadMessages[key] = (threadMessages[key] ?? [])
.appending(processedMessage?.messageInfo)
}
catch {
switch error {
// Ignore duplicate & selfSend message errors (and don't bother logging
// them as there will be a lot since we each service node duplicates messages)
case DatabaseError.SQLITE_CONSTRAINT_UNIQUE,
MessageReceiverError.duplicateMessage,
MessageReceiverError.duplicateControlMessage,
MessageReceiverError.selfSend:
break
messages
.compactMap { message -> ProcessedMessage? in
do {
return try Message.processRawReceivedMessage(db, rawMessage: message)
}
catch {
switch error {
// Ignore duplicate & selfSend message errors (and don't bother
// logging them as there will be a lot since we each service node
// duplicates messages)
case DatabaseError.SQLITE_CONSTRAINT_UNIQUE,
MessageReceiverError.duplicateMessage,
MessageReceiverError.duplicateControlMessage,
MessageReceiverError.selfSend:
break
// In the background ignore 'SQLITE_ABORT' (it generally means
// the BackgroundPoller has timed out
case DatabaseError.SQLITE_ABORT: break
default: SNLog("Failed to deserialize envelope due to error: \(error).")
}
default: SNLog("Failed to deserialize envelope due to error: \(error).")
return nil
}
}
}
threadMessages
.grouped { threadId, _, _ in (threadId ?? Message.nonThreadMessageId) }
.forEach { threadId, threadMessages in
let maybeJob: Job? = Job(
variant: .messageReceive,
behaviour: .runOnce,
threadId: threadId,
details: MessageReceiveJob.Details(
messages: threadMessages,
messages: threadMessages.map { $0.messageInfo },
isBackgroundPoll: true
)
)

34
SessionMessagingKit/Database/Models/Capability.swift

@ -59,3 +59,37 @@ public struct Capability: Codable, FetchableRecord, PersistableRecord, TableReco
self.isMissing = isMissing
}
}
extension Capability.Variant {
// MARK: - Codable
public init(from decoder: Decoder) throws {
let container: SingleValueDecodingContainer = try decoder.singleValueContainer()
let valueString: String = try container.decode(String.self)
// FIXME: Remove this code
// There was a point where we didn't have custom Codable handling for the Capability.Variant
// which resulted in the data being encoded into the database as a JSON dict - this code catches
// that case and extracts the standard string value so it can be processed the same as the
// "proper" custom Codable logic)
if valueString.starts(with: "{") {
self = Capability.Variant(
from: valueString
.replacingOccurrences(of: "\":{}}", with: "")
.replacingOccurrences(of: "\"}}", with: "")
.replacingOccurrences(of: "{\"unsupported\":{\"_0\":\"", with: "")
.replacingOccurrences(of: "{\"", with: "")
)
return
}
// FIXME: Remove this code ^^^
self = Capability.Variant(from: valueString)
}
public func encode(to encoder: Encoder) throws {
var container: SingleValueEncodingContainer = encoder.singleValueContainer()
try container.encode(rawValue)
}
}

52
SessionMessagingKit/Open Groups/Models/Capabilities.swift

@ -4,60 +4,14 @@ import Foundation
extension OpenGroupAPI {
public struct Capabilities: Codable, Equatable {
public enum Capability: Equatable, CaseIterable, Codable {
public static var allCases: [Capability] {
[.sogs, .blind]
}
case sogs
case blind
/// Fallback case if the capability isn't supported by this version of the app
case unsupported(String)
// MARK: - Convenience
public var rawValue: String {
switch self {
case .unsupported(let originalValue): return originalValue
default: return "\(self)"
}
}
// MARK: - Initialization
public init(from valueString: String) {
let maybeValue: Capability? = Capability.allCases.first { $0.rawValue == valueString }
self = (maybeValue ?? .unsupported(valueString))
}
}
public let capabilities: [Capability]
public let missing: [Capability]?
public let capabilities: [Capability.Variant]
public let missing: [Capability.Variant]?
// MARK: - Initialization
public init(capabilities: [Capability], missing: [Capability]? = nil) {
public init(capabilities: [Capability.Variant], missing: [Capability.Variant]? = nil) {
self.capabilities = capabilities
self.missing = missing
}
}
}
extension OpenGroupAPI.Capabilities.Capability {
// MARK: - Codable
public init(from decoder: Decoder) throws {
let container: SingleValueDecodingContainer = try decoder.singleValueContainer()
let valueString: String = try container.decode(String.self)
self = OpenGroupAPI.Capabilities.Capability(from: valueString)
}
public func encode(to encoder: Encoder) throws {
var container: SingleValueEncodingContainer = encoder.singleValueContainer()
try container.encode(rawValue)
}
}

3
SessionMessagingKit/Open Groups/Models/SOGSMessage.swift

@ -10,6 +10,7 @@ extension OpenGroupAPI {
case sender = "session_id"
case posted
case edited
case deleted
case seqNo = "seqno"
case whisper
case whisperMods = "whisper_mods"
@ -23,6 +24,7 @@ extension OpenGroupAPI {
public let sender: String?
public let posted: TimeInterval
public let edited: TimeInterval?
public let deleted: Bool?
public let seqNo: Int64
public let whisper: Bool
public let whisperMods: Bool
@ -79,6 +81,7 @@ extension OpenGroupAPI.Message {
sender: try? container.decode(String.self, forKey: .sender),
posted: try container.decode(TimeInterval.self, forKey: .posted),
edited: try? container.decode(TimeInterval.self, forKey: .edited),
deleted: try? container.decode(Bool.self, forKey: .deleted),
seqNo: try container.decode(Int64.self, forKey: .seqNo),
whisper: ((try? container.decode(Bool.self, forKey: .whisper)) ?? false),
whisperMods: ((try? container.decode(Bool.self, forKey: .whisperMods)) ?? false),

15
SessionMessagingKit/Open Groups/OpenGroupManager.swift

@ -348,7 +348,7 @@ public final class OpenGroupManager: NSObject {
capabilities.capabilities.forEach { capability in
_ = try? Capability(
openGroupServer: server.lowercased(),
variant: Capability.Variant(from: capability.rawValue),
variant: capability,
isMissing: false
)
.saved(db)
@ -356,7 +356,7 @@ public final class OpenGroupManager: NSObject {
capabilities.missing?.forEach { capability in
_ = try? Capability(
openGroupServer: server.lowercased(),
variant: Capability.Variant(from: capability.rawValue),
variant: capability,
isMissing: true
)
.saved(db)
@ -499,9 +499,12 @@ public final class OpenGroupManager: NSObject {
}
let sortedMessages: [OpenGroupAPI.Message] = messages
.filter { $0.deleted != true }
.sorted { lhs, rhs in lhs.id < rhs.id }
let messageServerIdsToRemove: [Int64] = messages
.filter { $0.deleted == true }
.map { $0.id }
let seqNo: Int64? = sortedMessages.map { $0.seqNo }.max()
var messageServerIdsToRemove: [UInt64] = []
// Update the 'openGroupSequenceNumber' value (Note: SOGS V4 uses the 'seqNo' instead of the 'serverId')
if let seqNo: Int64 = seqNo {
@ -515,11 +518,7 @@ public final class OpenGroupManager: NSObject {
guard
let base64EncodedString: String = message.base64EncodedData,
let data = Data(base64Encoded: base64EncodedString)
else {
// A message with no data has been deleted so add it to the list to remove
messageServerIdsToRemove.append(UInt64(message.id))
return
}
else { return }
do {
let processedMessage: ProcessedMessage? = try Message.processReceivedOpenGroupMessage(

7
SessionMessagingKit/Sending & Receiving/Message Handling/MessageReceiver+TypingIndicators.swift

@ -13,8 +13,7 @@ extension MessageReceiver {
switch message.kind {
case .started:
TypingIndicators.didStartTyping(
db,
let needsToStartTypingIndicator: Bool = TypingIndicators.didStartTypingNeedsToStart(
threadId: thread.id,
threadVariant: thread.variant,
threadIsMessageRequest: thread.isMessageRequest(db),
@ -22,6 +21,10 @@ extension MessageReceiver {
timestampMs: message.sentTimestamp.map { Int64($0) }
)
if needsToStartTypingIndicator {
TypingIndicators.start(db, threadId: thread.id, direction: .incoming)
}
case .stopped:
TypingIndicators.didStopTyping(db, threadId: thread.id, direction: .incoming)

41
SessionMessagingKit/Sending & Receiving/MessageSender.swift

@ -291,7 +291,7 @@ public final class MessageSender {
errorCount += 1
guard errorCount == promiseCount else { return } // Only error out if all promises failed
Storage.shared.write { db in
Storage.shared.read { db in
handleFailure(db, with: .other(error))
}
}
@ -300,7 +300,7 @@ public final class MessageSender {
.catch(on: DispatchQueue.global(qos: .default)) { error in
SNLog("Couldn't send message due to error: \(error).")
Storage.shared.write { db in
Storage.shared.read { db in
handleFailure(db, with: .other(error))
}
}
@ -447,7 +447,7 @@ public final class MessageSender {
}
}
.catch(on: DispatchQueue.global(qos: .default)) { error in
dependencies.storage.write { db in
dependencies.storage.read { db in
handleFailure(db, with: .other(error))
}
}
@ -557,7 +557,7 @@ public final class MessageSender {
}
}
.catch(on: DispatchQueue.global(qos: .default)) { error in
dependencies.storage.write { db in
dependencies.storage.read { db in
handleFailure(db, with: .other(error))
}
}
@ -652,15 +652,34 @@ public final class MessageSender {
with error: MessageSenderError,
interactionId: Int64?
) {
// Mark any "sending" recipients as "failed"
_ = try? RecipientState
// Check if we need to mark any "sending" recipients as "failed"
//
// Note: The 'db' could be either read-only or writeable so we determine
// if a change is required, and if so dispatch to a separate queue for the
// actual write
let rowIds: [Int64] = (try? RecipientState
.select(Column.rowID)
.filter(RecipientState.Columns.interactionId == interactionId)
.filter(RecipientState.Columns.state == RecipientState.State.sending)
.updateAll(
db,
RecipientState.Columns.state.set(to: RecipientState.State.failed),
RecipientState.Columns.mostRecentFailureText.set(to: error.localizedDescription)
)
.asRequest(of: Int64.self)
.fetchAll(db))
.defaulting(to: [])
guard !rowIds.isEmpty else { return }
// Need to dispatch to a different thread to prevent a potential db re-entrancy
// issue from occuring in some cases
DispatchQueue.global(qos: .background).async {
Storage.shared.write { db in
try RecipientState
.filter(rowIds.contains(Column.rowID))
.updateAll(
db,
RecipientState.Columns.state.set(to: RecipientState.State.failed),
RecipientState.Columns.mostRecentFailureText.set(to: error.localizedDescription)
)
}
}
}
// MARK: - Convenience

80
SessionMessagingKit/Sending & Receiving/Pollers/ClosedGroupPoller.swift

@ -152,6 +152,7 @@ public final class ClosedGroupPoller {
on queue: DispatchQueue = SessionSnodeKit.Threading.workQueue,
maxRetryCount: UInt = 0,
isBackgroundPoll: Bool = false,
isBackgroundPollValid: @escaping (() -> Bool) = { true },
poller: ClosedGroupPoller? = nil
) -> Promise<Void> {
let promise: Promise<Void> = SnodeAPI.getSwarm(for: groupPublicKey)
@ -160,9 +161,10 @@ public final class ClosedGroupPoller {
guard let snode = swarm.randomElement() else { return Promise(error: Error.insufficientSnodes) }
return attempt(maxRetryCount: maxRetryCount, recoveringOn: queue) {
guard isBackgroundPoll || poller?.isPolling.wrappedValue[groupPublicKey] == true else {
return Promise(error: Error.pollingCanceled)
}
guard
(isBackgroundPoll && isBackgroundPollValid()) ||
poller?.isPolling.wrappedValue[groupPublicKey] == true
else { return Promise(error: Error.pollingCanceled) }
let promises: [Promise<[SnodeReceivedMessage]>] = {
if SnodeAPI.hardfork >= 19 && SnodeAPI.softfork >= 1 {
@ -181,9 +183,13 @@ public final class ClosedGroupPoller {
return when(resolved: promises)
.then(on: queue) { messageResults -> Promise<Void> in
guard isBackgroundPoll || poller?.isPolling.wrappedValue[groupPublicKey] == true else { return Promise.value(()) }
guard
(isBackgroundPoll && isBackgroundPollValid()) ||
poller?.isPolling.wrappedValue[groupPublicKey] == true
else { return Promise.value(()) }
var promises: [Promise<Void>] = []
var jobToRun: Job? = nil
let allMessages: [SnodeReceivedMessage] = messageResults
.reduce([]) { result, next in
switch next {
@ -192,8 +198,16 @@ public final class ClosedGroupPoller {
}
}
var messageCount: Int = 0
let totalMessagesCount: Int = allMessages.count
// No need to do anything if there are no messages
guard !allMessages.isEmpty else {
if !isBackgroundPoll {
SNLog("Received no new messages in closed group with public key: \(groupPublicKey)")
}
return Promise.value(())
}
// Otherwise process the messages and add them to the queue for handling
Storage.shared.write { db in
let processedMessages: [ProcessedMessage] = allMessages
.compactMap { message -> ProcessedMessage? in
@ -209,6 +223,14 @@ public final class ClosedGroupPoller {
MessageReceiverError.duplicateControlMessage,
MessageReceiverError.selfSend:
break
// In the background ignore 'SQLITE_ABORT' (it generally means
// the BackgroundPoller has timed out
case DatabaseError.SQLITE_ABORT:
guard !isBackgroundPoll else { break }
SNLog("Failed to the database being suspended (running in background with no background task).")
break
default: SNLog("Failed to deserialize envelope due to error: \(error).")
}
@ -219,7 +241,7 @@ public final class ClosedGroupPoller {
messageCount = processedMessages.count
let jobToRun: Job? = Job(
jobToRun = Job(
variant: .messageReceive,
behaviour: .runOnce,
threadId: groupPublicKey,
@ -232,35 +254,29 @@ public final class ClosedGroupPoller {
// If we are force-polling then add to the JobRunner so they are persistent and will retry on
// the next app run if they fail but don't let them auto-start
JobRunner.add(db, job: jobToRun, canStartJob: !isBackgroundPoll)
}
if isBackgroundPoll {
// We want to try to handle the receive jobs immediately in the background
if isBackgroundPoll {
promises = promises.appending(
jobToRun.map { job -> Promise<Void> in
let (promise, seal) = Promise<Void>.pending()
// Note: In the background we just want jobs to fail silently
MessageReceiveJob.run(
job,
queue: queue,
success: { _, _ in seal.fulfill(()) },
failure: { _, _, _ in seal.fulfill(()) },
deferred: { _ in seal.fulfill(()) }
)
promises = promises.appending(
jobToRun.map { job -> Promise<Void> in
let (promise, seal) = Promise<Void>.pending()
// Note: In the background we just want jobs to fail silently
MessageReceiveJob.run(
job,
queue: queue,
success: { _, _ in seal.fulfill(()) },
failure: { _, _, _ in seal.fulfill(()) },
deferred: { _ in seal.fulfill(()) }
)
return promise
}
)
}
return promise
}
)
}
if !isBackgroundPoll {
if totalMessagesCount > 0 {
SNLog("Received \(messageCount) new message\(messageCount == 1 ? "" : "s") in closed group with public key: \(groupPublicKey) (duplicates: \(totalMessagesCount - messageCount))")
}
else {
SNLog("Received no new messages in closed group with public key: \(groupPublicKey)")
}
else {
SNLog("Received \(messageCount) new message\(messageCount == 1 ? "" : "s") in closed group with public key: \(groupPublicKey) (duplicates: \(allMessages.count - messageCount))")
}
return when(fulfilled: promises)

242
SessionMessagingKit/Sending & Receiving/Pollers/OpenGroupPoller.swift

@ -8,6 +8,8 @@ import SessionUtilitiesKit
extension OpenGroupAPI {
public final class Poller {
typealias PollResponse = [OpenGroupAPI.Endpoint: (info: OnionRequestResponseInfoType, data: Codable?)]
private let server: String
private var timer: Timer? = nil
private var hasStarted = false
@ -71,6 +73,7 @@ extension OpenGroupAPI {
@discardableResult
public func poll(
isBackgroundPoll: Bool,
isBackgroundPollerValid: @escaping (() -> Bool) = { true },
isPostCapabilitiesRetry: Bool,
using dependencies: OpenGroupManager.OGMDependencies = OpenGroupManager.OGMDependencies()
) -> Promise<Void> {
@ -83,8 +86,14 @@ extension OpenGroupAPI {
Threading.pollerQueue.async {
dependencies.storage
.read { db in
OpenGroupAPI
.read { db -> Promise<(Int64, PollResponse)> in
let failureCount: Int64 = (try? OpenGroup
.select(max(OpenGroup.Columns.pollFailureCount))
.asRequest(of: Int64.self)
.fetchOne(db))
.defaulting(to: 0)
return OpenGroupAPI
.poll(
db,
server: server,
@ -95,10 +104,24 @@ extension OpenGroupAPI {
),
using: dependencies
)
.map(on: OpenGroupAPI.workQueue) { (failureCount, $0) }
}
.done(on: OpenGroupAPI.workQueue) { [weak self] response in
.done(on: OpenGroupAPI.workQueue) { [weak self] failureCount, response in
guard !isBackgroundPoll || isBackgroundPollerValid() else {
// If this was a background poll and the background poll is no longer valid
// then just stop
self?.isPolling = false
seal.fulfill(())
return
}
self?.isPolling = false
self?.handlePollResponse(response, isBackgroundPoll: isBackgroundPoll, using: dependencies)
self?.handlePollResponse(
response,
failureCount: failureCount,
isBackgroundPoll: isBackgroundPoll,
using: dependencies
)
dependencies.mutableCache.mutate { cache in
cache.hasPerformedInitialPoll[server] = true
@ -106,17 +129,18 @@ extension OpenGroupAPI {
UserDefaults.standard[.lastOpen] = Date()
}
// Reset the failure count
Storage.shared.writeAsync { db in
try OpenGroup
.filter(OpenGroup.Columns.server == server)
.updateAll(db, OpenGroup.Columns.pollFailureCount.set(to: 0))
}
SNLog("Open group polling finished for \(server).")
seal.fulfill(())
}
.catch(on: OpenGroupAPI.workQueue) { [weak self] error in
guard !isBackgroundPoll || isBackgroundPollerValid() else {
// If this was a background poll and the background poll is no longer valid
// then just stop
self?.isPolling = false
seal.fulfill(())
return
}
// If we are retrying then the error is being handled so no need to continue (this
// method will always resolve)
self?.updateCapabilitiesAndRetryIfNeeded(
@ -141,7 +165,10 @@ extension OpenGroupAPI {
Storage.shared.writeAsync { db in
try OpenGroup
.filter(OpenGroup.Columns.server == server)
.updateAll(db, OpenGroup.Columns.pollFailureCount.set(to: (pollFailureCount + 1)))
.updateAll(
db,
OpenGroup.Columns.pollFailureCount.set(to: (pollFailureCount + 1))
)
}
SNLog("Open group polling failed due to error: \(error). Setting failure count to \(pollFailureCount).")
@ -221,50 +248,47 @@ extension OpenGroupAPI {
return promise
}
private func handlePollResponse(_ response: [OpenGroupAPI.Endpoint: (info: OnionRequestResponseInfoType, data: Codable?)], isBackgroundPoll: Bool, using dependencies: OpenGroupManager.OGMDependencies = OpenGroupManager.OGMDependencies()) {
private func handlePollResponse(
_ response: PollResponse,
failureCount: Int64,
isBackgroundPoll: Bool,
using dependencies: OpenGroupManager.OGMDependencies = OpenGroupManager.OGMDependencies()
) {
let server: String = self.server
dependencies.storage.write { db in
try response.forEach { endpoint, endpointResponse in
let validResponses: PollResponse = response
.filter { endpoint, endpointResponse in
switch endpoint {
case .capabilities:
guard let responseData: BatchSubResponse<Capabilities> = endpointResponse.data as? BatchSubResponse<Capabilities>, let responseBody: Capabilities = responseData.body else {
guard (endpointResponse.data as? BatchSubResponse<Capabilities>)?.body != nil else {
SNLog("Open group polling failed due to invalid capability data.")
return
return false
}
OpenGroupManager.handleCapabilities(
db,
capabilities: responseBody,
on: server
)
return true
case .roomPollInfo(let roomToken, _):
guard let responseData: BatchSubResponse<RoomPollInfo> = endpointResponse.data as? BatchSubResponse<RoomPollInfo>, let responseBody: RoomPollInfo = responseData.body else {
guard (endpointResponse.data as? BatchSubResponse<RoomPollInfo>)?.body != nil else {
switch (endpointResponse.data as? BatchSubResponse<RoomPollInfo>)?.code {
case 404: SNLog("Open group polling failed to retrieve info for unknown room '\(roomToken)'.")
default: SNLog("Open group polling failed due to invalid room info data.")
}
return
return false
}
try OpenGroupManager.handlePollInfo(
db,
pollInfo: responseBody,
publicKey: nil,
for: roomToken,
on: server,
dependencies: dependencies
)
return true
case .roomMessagesRecent(let roomToken), .roomMessagesBefore(let roomToken, _), .roomMessagesSince(let roomToken, _):
guard let responseData: BatchSubResponse<[Failable<Message>]> = endpointResponse.data as? BatchSubResponse<[Failable<Message>]>, let responseBody: [Failable<Message>] = responseData.body else {
guard
let responseData: BatchSubResponse<[Failable<Message>]> = endpointResponse.data as? BatchSubResponse<[Failable<Message>]>,
let responseBody: [Failable<Message>] = responseData.body
else {
switch (endpointResponse.data as? BatchSubResponse<[Failable<Message>]>)?.code {
case 404: SNLog("Open group polling failed to retrieve messages for unknown room '\(roomToken)'.")
default: SNLog("Open group polling failed due to invalid messages data.")
}
return
return false
}
let successfulMessages: [Message] = responseBody.compactMap { $0.value }
if successfulMessages.count != responseBody.count {
@ -273,9 +297,147 @@ extension OpenGroupAPI {
SNLog("Dropped \(droppedCount) invalid open group message\(droppedCount == 1 ? "" : "s").")
}
return !successfulMessages.isEmpty
case .inbox, .inboxSince, .outbox, .outboxSince:
guard
let responseData: BatchSubResponse<[DirectMessage]?> = endpointResponse.data as? BatchSubResponse<[DirectMessage]?>,
!responseData.failedToParseBody
else {
SNLog("Open group polling failed due to invalid inbox/outbox data.")
return false
}
// Double optional because the server can return a `304` with an empty body
let messages: [OpenGroupAPI.DirectMessage] = ((responseData.body ?? []) ?? [])
return !messages.isEmpty
default: return false // No custom handling needed
}
}
// If there are no remaining 'validResponses' and there hasn't been a failure then there is
// no need to do anything else
guard !validResponses.isEmpty || failureCount != 0 else { return }
// Retrieve the current capability & group info to check if anything changed
let rooms: [String] = validResponses
.keys
.compactMap { endpoint -> String? in
switch endpoint {
case .roomPollInfo(let roomToken, _): return roomToken
default: return nil
}
}
let currentInfo: (capabilities: Capabilities, groups: [OpenGroup])? = dependencies.storage.read { db in
let allCapabilities: [Capability] = try Capability
.filter(Capability.Columns.openGroupServer == server)
.fetchAll(db)
let capabilities: Capabilities = Capabilities(
capabilities: allCapabilities
.filter { !$0.isMissing }
.map { $0.variant },
missing: {
let missingCapabilities: [Capability.Variant] = allCapabilities
.filter { $0.isMissing }
.map { $0.variant }
return (missingCapabilities.isEmpty ? nil : missingCapabilities)
}()
)
let openGroupIds: [String] = rooms
.map { OpenGroup.idFor(roomToken: $0, server: server) }
let groups: [OpenGroup] = try OpenGroup
.filter(ids: openGroupIds)
.fetchAll(db)
return (capabilities, groups)
}
let changedResponses: PollResponse = validResponses
.filter { endpoint, endpointResponse in
switch endpoint {
case .capabilities:
guard
let responseData: BatchSubResponse<Capabilities> = endpointResponse.data as? BatchSubResponse<Capabilities>,
let responseBody: Capabilities = responseData.body
else { return false }
return (responseBody != currentInfo?.capabilities)
case .roomPollInfo(let roomToken, _):
guard
let responseData: BatchSubResponse<RoomPollInfo> = endpointResponse.data as? BatchSubResponse<RoomPollInfo>,
let responseBody: RoomPollInfo = responseData.body
else { return false }
guard let existingOpenGroup: OpenGroup = currentInfo?.groups.first(where: { $0.roomToken == roomToken }) else {
return true
}
// Note: This might need to be updated in the future when we start tracking
// user permissions if changes to permissions don't trigger a change to
// the 'infoUpdates'
return (
responseBody.activeUsers != existingOpenGroup.userCount || (
responseBody.details != nil &&
responseBody.details?.infoUpdates != existingOpenGroup.infoUpdates
)
)
default: return true
}
}
// If there are no 'changedResponses' and there hasn't been a failure then there is
// no need to do anything else
guard !changedResponses.isEmpty || failureCount != 0 else { return }
dependencies.storage.write { db in
// Reset the failure count
if failureCount > 0 {
try OpenGroup
.filter(OpenGroup.Columns.server == server)
.updateAll(db, OpenGroup.Columns.pollFailureCount.set(to: 0))
}
try changedResponses.forEach { endpoint, endpointResponse in
switch endpoint {
case .capabilities:
guard
let responseData: BatchSubResponse<Capabilities> = endpointResponse.data as? BatchSubResponse<Capabilities>,
let responseBody: Capabilities = responseData.body
else { return }
OpenGroupManager.handleCapabilities(
db,
capabilities: responseBody,
on: server
)
case .roomPollInfo(let roomToken, _):
guard
let responseData: BatchSubResponse<RoomPollInfo> = endpointResponse.data as? BatchSubResponse<RoomPollInfo>,
let responseBody: RoomPollInfo = responseData.body
else { return }
try OpenGroupManager.handlePollInfo(
db,
pollInfo: responseBody,
publicKey: nil,
for: roomToken,
on: server,
dependencies: dependencies
)
case .roomMessagesRecent(let roomToken), .roomMessagesBefore(let roomToken, _), .roomMessagesSince(let roomToken, _):
guard
let responseData: BatchSubResponse<[Failable<Message>]> = endpointResponse.data as? BatchSubResponse<[Failable<Message>]>,
let responseBody: [Failable<Message>] = responseData.body
else { return }
OpenGroupManager.handleMessages(
db,
messages: successfulMessages,
messages: responseBody.compactMap { $0.value },
for: roomToken,
on: server,
isBackgroundPoll: isBackgroundPoll,
@ -283,10 +445,10 @@ extension OpenGroupAPI {
)
case .inbox, .inboxSince, .outbox, .outboxSince:
guard let responseData: BatchSubResponse<[DirectMessage]?> = endpointResponse.data as? BatchSubResponse<[DirectMessage]?>, !responseData.failedToParseBody else {
SNLog("Open group polling failed due to invalid inbox/outbox data.")
return
}
guard
let responseData: BatchSubResponse<[DirectMessage]?> = endpointResponse.data as? BatchSubResponse<[DirectMessage]?>,
!responseData.failedToParseBody
else { return }
// Double optional because the server can return a `304` with an empty body