keejef
/
session-ios
mirror of https://github.com/oxen-io/session-ios
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #156 from loki-project/reliability

Browse Source 
Fix Retrying
pull/158/head
Niels Andriesse 5 years ago committed by GitHub
parent 664717b1d0 838688355b
commit 5cf82fffdf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 157 additions and 131 deletions
Show Stats Download Patch File Download Diff File

82
SignalServiceKit/src/Loki/API/HTTP/HTTP.swift
Unescape Escape View File

@ -41,53 +41,53 @@ internal enum HTTP {
// MARK: Main
internal static func execute(_ verb: Verb, _ url: String, parameters: JSON? = nil, timeout: TimeInterval = HTTP.timeout) -> Promise<JSON> {
return Promise<JSON> { seal in
var request = URLRequest(url: URL(string: url)!)
request.httpMethod = verb.rawValue
if let parameters = parameters {
do {
guard JSONSerialization.isValidJSONObject(parameters) else { return seal.reject(Error.invalidJSON) }
request.httpBody = try JSONSerialization.data(withJSONObject: parameters, options: [])
} catch (let error) {
return seal.reject(error)
}
var request = URLRequest(url: URL(string: url)!)
request.httpMethod = verb.rawValue
if let parameters = parameters {
do {
guard JSONSerialization.isValidJSONObject(parameters) else { return Promise(error: Error.invalidJSON) }
request.httpBody = try JSONSerialization.data(withJSONObject: parameters, options: [])
} catch (let error) {
return Promise(error: error)
}
request.timeoutInterval = timeout
let task = urlSession.dataTask(with: request) { data, response, error in
guard let data = data, let response = response as? HTTPURLResponse else {
if let error = error {
print("[Loki] \(verb.rawValue) request to \(url) failed due to error: \(error).")
} else {
print("[Loki] \(verb.rawValue) request to \(url) failed.")
}
// Override the actual error so that we can correctly catch failed requests in sendOnionRequest(invoking:on:with:)
return seal.reject(Error.httpRequestFailed(statusCode: 0, json: nil))
}
}
request.timeoutInterval = timeout
let (promise, seal) = Promise<JSON>.pending()
let task = urlSession.dataTask(with: request) { data, response, error in
guard let data = data, let response = response as? HTTPURLResponse else {
if let error = error {
print("[Loki] \(verb.rawValue) request to \(url) failed due to error: \(error).")
// Override the actual error so that we can correctly catch failed requests in sendOnionRequest(invoking:on:with:)
return seal.reject(Error.httpRequestFailed(statusCode: 0, json: nil))
}
let statusCode = UInt(response.statusCode)
var json: JSON? = nil
if let j = try? JSONSerialization.jsonObject(with: data, options: []) as? JSON {
json = j
} else if let result = String(data: data, encoding: .utf8) {
json = [ "result" : result ]
}
guard 200...299 ~= statusCode else {
let jsonDescription = json?.prettifiedDescription ?? "no debugging info provided"
print("[Loki] \(verb.rawValue) request to \(url) failed with status code: \(statusCode) (\(jsonDescription)).")
return seal.reject(Error.httpRequestFailed(statusCode: statusCode, json: json))
}
if let json = json {
seal.fulfill(json)
} else {
print("[Loki] Couldn't parse JSON returned by \(verb.rawValue) request to \(url).")
return seal.reject(Error.invalidJSON)
print("[Loki] \(verb.rawValue) request to \(url) failed.")
}
// Override the actual error so that we can correctly catch failed requests in sendOnionRequest(invoking:on:with:)
return seal.reject(Error.httpRequestFailed(statusCode: 0, json: nil))
}
if let error = error {
print("[Loki] \(verb.rawValue) request to \(url) failed due to error: \(error).")
// Override the actual error so that we can correctly catch failed requests in sendOnionRequest(invoking:on:with:)
return seal.reject(Error.httpRequestFailed(statusCode: 0, json: nil))
}
let statusCode = UInt(response.statusCode)
var json: JSON? = nil
if let j = try? JSONSerialization.jsonObject(with: data, options: []) as? JSON {
json = j
} else if let result = String(data: data, encoding: .utf8) {
json = [ "result" : result ]
}
guard 200...299 ~= statusCode else {
let jsonDescription = json?.prettifiedDescription ?? "no debugging info provided"
print("[Loki] \(verb.rawValue) request to \(url) failed with status code: \(statusCode) (\(jsonDescription)).")
return seal.reject(Error.httpRequestFailed(statusCode: statusCode, json: json))
}
if let json = json {
seal.fulfill(json)
} else {
print("[Loki] Couldn't parse JSON returned by \(verb.rawValue) request to \(url).")
return seal.reject(Error.invalidJSON)
}
task.resume()
}
task.resume()
return promise
}
}

29
SignalServiceKit/src/Loki/API/LokiAPI+SwarmAPI.swift
Unescape Escape View File

@ -55,21 +55,22 @@ public extension LokiAPI {
]
print("[Loki] Populating snode pool using: \(target).")
let (promise, seal) = Promise<LokiAPITarget>.pending()
let queue = workQueue
HTTP.execute(.post, url, parameters: parameters).map(on: queue) { json in
guard let intermediate = json["result"] as? JSON, let rawTargets = intermediate["service_node_states"] as? [JSON] else { throw LokiAPIError.randomSnodePoolUpdatingFailed }
randomSnodePool = try Set(rawTargets.flatMap { rawTarget in
guard let address = rawTarget["public_ip"] as? String, let port = rawTarget["storage_port"] as? Int, let ed25519PublicKey = rawTarget["pubkey_ed25519"] as? String, let x25519PublicKey = rawTarget["pubkey_x25519"] as? String, address != "0.0.0.0" else {
print("[Loki] Failed to parse target from: \(rawTarget).")
return nil
}
return LokiAPITarget(address: "https://\(address)", port: UInt16(port), publicKeySet: LokiAPITarget.KeySet(ed25519Key: ed25519PublicKey, x25519Key: x25519PublicKey))
})
// randomElement() uses the system's default random generator, which is cryptographically secure
return randomSnodePool.randomElement()!
}.retryingIfNeeded(maxRetryCount: 4).done(on: queue) { snode in
attempt(maxRetryCount: 4, recoveringOn: workQueue) {
HTTP.execute(.post, url, parameters: parameters).map(on: workQueue) { json in
guard let intermediate = json["result"] as? JSON, let rawTargets = intermediate["service_node_states"] as? [JSON] else { throw LokiAPIError.randomSnodePoolUpdatingFailed }
randomSnodePool = try Set(rawTargets.flatMap { rawTarget in
guard let address = rawTarget["public_ip"] as? String, let port = rawTarget["storage_port"] as? Int, let ed25519PublicKey = rawTarget["pubkey_ed25519"] as? String, let x25519PublicKey = rawTarget["pubkey_x25519"] as? String, address != "0.0.0.0" else {
print("[Loki] Failed to parse target from: \(rawTarget).")
return nil
}
return LokiAPITarget(address: "https://\(address)", port: UInt16(port), publicKeySet: LokiAPITarget.KeySet(ed25519Key: ed25519PublicKey, x25519Key: x25519PublicKey))
})
// randomElement() uses the system's default random generator, which is cryptographically secure
return randomSnodePool.randomElement()!
}
}.done(on: workQueue) { snode in
seal.fulfill(snode)
}.catch(on: queue) { error in
}.catch(on: workQueue) { error in
print("[Loki] Failed to contact seed node at: \(target).")
seal.reject(error)
}

25
SignalServiceKit/src/Loki/API/LokiAPI.swift
Unescape Escape View File

@ -122,9 +122,11 @@ public final class LokiAPI : NSObject {
// MARK: Public API
public static func getMessages() -> Promise<Set<MessageListPromise>> {
return getTargetSnodes(for: userHexEncodedPublicKey).mapValues { targetSnode in
return getRawMessages(from: targetSnode, usingLongPolling: false).map { parseRawMessagesResponse($0, from: targetSnode) }
}.map { Set($0) }.retryingIfNeeded(maxRetryCount: maxRetryCount)
return attempt(maxRetryCount: maxRetryCount, recoveringOn: workQueue) {
getTargetSnodes(for: userHexEncodedPublicKey).mapValues { targetSnode in
getRawMessages(from: targetSnode, usingLongPolling: false).map { parseRawMessagesResponse($0, from: targetSnode) }
}.map { Set($0) }
}
}
public static func getDestinations(for hexEncodedPublicKey: String) -> Promise<[Destination]> {
@ -189,16 +191,18 @@ public final class LokiAPI : NSObject {
let destination = lokiMessage.destination
func sendLokiMessage(_ lokiMessage: LokiMessage, to target: LokiAPITarget) -> RawResponsePromise {
let parameters = lokiMessage.toJSON()
return invoke(.sendMessage, on: target, associatedWith: destination, parameters: parameters)
return attempt(maxRetryCount: maxRetryCount, recoveringOn: workQueue) {
invoke(.sendMessage, on: target, associatedWith: destination, parameters: parameters)
}
}
func sendLokiMessageUsingSwarmAPI() -> Promise<Set<RawResponsePromise>> {
notificationCenter.post(name: .calculatingPoW, object: NSNumber(value: signalMessage.timestamp))
return lokiMessage.calculatePoW().then { lokiMessageWithPoW -> Promise<Set<RawResponsePromise>> in
notificationCenter.post(name: .routing, object: NSNumber(value: signalMessage.timestamp))
return getTargetSnodes(for: destination).map { swarm in
return Set(swarm.map { target in
return getTargetSnodes(for: destination).map { snodes in
return Set(snodes.map { snode in
notificationCenter.post(name: .messageSending, object: NSNumber(value: signalMessage.timestamp))
return sendLokiMessage(lokiMessageWithPoW, to: target).map { rawResponse in
return sendLokiMessage(lokiMessageWithPoW, to: snode).map { rawResponse in
if let json = rawResponse as? JSON, let powDifficulty = json["difficulty"] as? Int {
guard powDifficulty != LokiAPI.powDifficulty else { return rawResponse }
print("[Loki] Setting proof of work difficulty to \(powDifficulty).")
@ -207,14 +211,15 @@ public final class LokiAPI : NSObject {
print("[Loki] Failed to update proof of work difficulty from: \(rawResponse).")
}
return rawResponse
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
}
})
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
}
}
}
if let peer = LokiP2PAPI.getInfo(for: destination), (lokiMessage.isPing || peer.isOnline) {
let target = LokiAPITarget(address: peer.address, port: peer.port, publicKeySet: nil)
return Promise.value([ target ]).mapValues { sendLokiMessage(lokiMessage, to: $0) }.map { Set($0) }.retryingIfNeeded(maxRetryCount: maxRetryCount).get { _ in
// TODO: Retrying
return Promise.value([ target ]).mapValues { sendLokiMessage(lokiMessage, to: $0) }.map { Set($0) }.get { _ in
LokiP2PAPI.markOnline(destination)
onP2PSuccess()
}.recover { error -> Promise<Set<RawResponsePromise>> in

4
SignalServiceKit/src/Loki/API/LokiFileServerAPI.swift
Unescape Escape View File

@ -101,7 +101,9 @@ public final class LokiFileServerAPI : LokiDotNetAPI {
let url = URL(string: "\(server)/users/me")!
let request = TSRequest(url: url, method: "PATCH", parameters: parameters)
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).map { _ in }.retryingIfNeeded(maxRetryCount: 8).recover { error in
return attempt(maxRetryCount: 8, recoveringOn: LokiAPI.workQueue) {
LokiFileServerProxy(for: server).perform(request).map { _ in }
}.recover { error in
print("Couldn't update device links due to error: \(error).")
throw error
}

5
SignalServiceKit/src/Loki/API/Onion Requests/OnionRequestAPI.swift
Unescape Escape View File

@ -145,6 +145,10 @@ internal enum OnionRequestAPI {
paths = paths.filter { !$0.contains(snode) }
}
private static func dropGuardSnode(_ snode: LokiAPITarget) {
guardSnodes = guardSnodes.filter { $0 != snode }
}
/// Builds an onion around `payload` and returns the result.
private static func buildOnion(around payload: JSON, targetedAt snode: LokiAPITarget) -> Promise<OnionBuildingResult> {
var guardSnode: LokiAPITarget!
@ -221,6 +225,7 @@ internal enum OnionRequestAPI {
promise.catch(on: LokiAPI.workQueue) { error in // Must be invoked on LokiAPI.workQueue
guard case HTTP.Error.httpRequestFailed(_, _) = error else { return }
dropPath(containing: guardSnode) // A snode in the path is bad; retry with a different path
dropGuardSnode(guardSnode)
}
promise.handlingErrorsIfNeeded(forTargetSnode: snode, associatedWith: hexEncodedPublicKey)
return promise

121
SignalServiceKit/src/Loki/API/Public Chats/LokiPublicChatAPI.swift
Unescape Escape View File

@ -155,11 +155,13 @@ public final class LokiPublicChatAPI : LokiDotNetAPI {
}
public static func sendMessage(_ message: LokiPublicChatMessage, to channel: UInt64, on server: String) -> Promise<LokiPublicChatMessage> {
return Promise<LokiPublicChatMessage> { [privateKey = userKeyPair.privateKey] seal in
DispatchQueue.global().async {
guard let signedMessage = message.sign(with: privateKey) else { return seal.reject(LokiDotNetAPIError.signingFailed) }
print("[Loki] Sending message to public chat channel with ID: \(channel) on server: \(server).")
let (promise, seal) = Promise<LokiPublicChatMessage>.pending()
let queue = DispatchQueue.global()
queue.async { [privateKey = userKeyPair.privateKey] in
guard let signedMessage = message.sign(with: privateKey) else { return seal.reject(LokiDotNetAPIError.signingFailed) }
attempt(maxRetryCount: maxRetryCount, recoveringOn: queue) {
getAuthToken(for: server).then { token -> Promise<LokiPublicChatMessage> in
print("[Loki] Sending message to public chat channel with ID: \(channel) on server: \(server).")
let url = URL(string: "\(server)/channels/\(channel)/messages")!
let parameters = signedMessage.toJSON()
let request = TSRequest(url: url, method: "POST", parameters: parameters)
@ -183,13 +185,14 @@ public final class LokiPublicChatAPI : LokiDotNetAPI {
storage.dbReadWriteConnection.removeObject(forKey: server, inCollection: authTokenCollection)
}
throw error
}.retryingIfNeeded(maxRetryCount: maxRetryCount).done { message in
seal.fulfill(message)
}.catch { error in
seal.reject(error)
}
}.done { message in
seal.fulfill(message)
}.catch { error in
seal.reject(error)
}
}
return promise
}
public static func getDeletedMessageServerIDs(for channel: UInt64, on server: String) -> Promise<[UInt64]> {
@ -220,16 +223,18 @@ public final class LokiPublicChatAPI : LokiDotNetAPI {
}
public static func deleteMessage(with messageID: UInt, for channel: UInt64, on server: String, isSentByUser: Bool) -> Promise<Void> {
return getAuthToken(for: server).then { token -> Promise<Void> in
let isModerationRequest = !isSentByUser
print("[Loki] Deleting message with ID: \(messageID) for public chat channel with ID: \(channel) on server: \(server) (isModerationRequest = \(isModerationRequest)).")
let urlAsString = isSentByUser ? "\(server)/channels/\(channel)/messages/\(messageID)" : "\(server)/loki/v1/moderation/message/\(messageID)"
let url = URL(string: urlAsString)!
let request = TSRequest(url: url, method: "DELETE", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Deleted message with ID: \(messageID) on server: \(server).")
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
return attempt(maxRetryCount: maxRetryCount, recoveringOn: DispatchQueue.global()) {
getAuthToken(for: server).then { token -> Promise<Void> in
let isModerationRequest = !isSentByUser
print("[Loki] Deleting message with ID: \(messageID) for public chat channel with ID: \(channel) on server: \(server) (isModerationRequest = \(isModerationRequest)).")
let urlAsString = isSentByUser ? "\(server)/channels/\(channel)/messages/\(messageID)" : "\(server)/loki/v1/moderation/message/\(messageID)"
let url = URL(string: urlAsString)!
let request = TSRequest(url: url, method: "DELETE", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Deleted message with ID: \(messageID) on server: \(server).")
}
}
}
}
@ -252,24 +257,28 @@ public final class LokiPublicChatAPI : LokiDotNetAPI {
}
public static func join(_ channel: UInt64, on server: String) -> Promise<Void> {
return getAuthToken(for: server).then { token -> Promise<Void> in
let url = URL(string: "\(server)/channels/\(channel)/subscribe")!
let request = TSRequest(url: url, method: "POST", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Joined channel with ID: \(channel) on server: \(server).")
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
return attempt(maxRetryCount: maxRetryCount, recoveringOn: DispatchQueue.global()) {
getAuthToken(for: server).then { token -> Promise<Void> in
let url = URL(string: "\(server)/channels/\(channel)/subscribe")!
let request = TSRequest(url: url, method: "POST", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Joined channel with ID: \(channel) on server: \(server).")
}
}
}
}
public static func leave(_ channel: UInt64, on server: String) -> Promise<Void> {
return getAuthToken(for: server).then { token -> Promise<Void> in
let url = URL(string: "\(server)/channels/\(channel)/subscribe")!
let request = TSRequest(url: url, method: "DELETE", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Left channel with ID: \(channel) on server: \(server).")
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
return attempt(maxRetryCount: maxRetryCount, recoveringOn: DispatchQueue.global()) {
getAuthToken(for: server).then { token -> Promise<Void> in
let url = URL(string: "\(server)/channels/\(channel)/subscribe")!
let request = TSRequest(url: url, method: "DELETE", parameters: [:])
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).done { result -> Void in
print("[Loki] Left channel with ID: \(channel) on server: \(server).")
}
}
}
}
@ -328,34 +337,38 @@ public final class LokiPublicChatAPI : LokiDotNetAPI {
public static func setDisplayName(to newDisplayName: String?, on server: String) -> Promise<Void> {
print("[Loki] Updating display name on server: \(server).")
return getAuthToken(for: server).then { token -> Promise<Void> in
let parameters: JSON = [ "name" : (newDisplayName ?? "") ]
let url = URL(string: "\(server)/users/me")!
let request = TSRequest(url: url, method: "PATCH", parameters: parameters)
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).map { _ in }.recover { error in
print("Couldn't update display name due to error: \(error).")
throw error
return attempt(maxRetryCount: maxRetryCount, recoveringOn: DispatchQueue.global()) {
getAuthToken(for: server).then { token -> Promise<Void> in
let parameters: JSON = [ "name" : (newDisplayName ?? "") ]
let url = URL(string: "\(server)/users/me")!
let request = TSRequest(url: url, method: "PATCH", parameters: parameters)
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).map { _ in }.recover { error in
print("Couldn't update display name due to error: \(error).")
throw error
}
}
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
}
}
public static func setProfilePictureURL(to url: String?, using profileKey: Data, on server: String) -> Promise<Void> {
print("[Loki] Updating profile picture on server: \(server).")
return getAuthToken(for: server).then { token -> Promise<Void> in
var annotation: JSON = [ "type" : profilePictureType ]
if let url = url {
annotation["value"] = [ "profileKey" : profileKey.base64EncodedString(), "url" : url ]
}
let parameters: JSON = [ "annotations" : [ annotation ] ]
let url = URL(string: "\(server)/users/me")!
let request = TSRequest(url: url, method: "PATCH", parameters: parameters)
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).map { _ in }.recover { error in
print("[Loki] Couldn't update profile picture due to error: \(error).")
throw error
return attempt(maxRetryCount: maxRetryCount, recoveringOn: DispatchQueue.global()) {
getAuthToken(for: server).then { token -> Promise<Void> in
var annotation: JSON = [ "type" : profilePictureType ]
if let url = url {
annotation["value"] = [ "profileKey" : profileKey.base64EncodedString(), "url" : url ]
}
let parameters: JSON = [ "annotations" : [ annotation ] ]
let url = URL(string: "\(server)/users/me")!
let request = TSRequest(url: url, method: "PATCH", parameters: parameters)
request.allHTTPHeaderFields = [ "Content-Type" : "application/json", "Authorization" : "Bearer \(token)" ]
return LokiFileServerProxy(for: server).perform(request).map { _ in }.recover { error in
print("[Loki] Couldn't update profile picture due to error: \(error).")
throw error
}
}
}.retryingIfNeeded(maxRetryCount: maxRetryCount)
}
}
public static func getInfo(for channel: UInt64, on server: String) -> Promise<LokiPublicChatInfo> {

22
SignalServiceKit/src/Loki/Utilities/Promise+Retrying.swift
Unescape Escape View File

@ -1,16 +1,16 @@
import PromiseKit
internal extension Promise {
internal func retryingIfNeeded(maxRetryCount: UInt) -> Promise<T> {
var retryCount = 0
func retryIfNeeded() -> Promise<T> {
return recover(on: DispatchQueue.global()) { error -> Promise<T> in
guard retryCount != maxRetryCount else { throw error }
retryCount += 1
return retryIfNeeded()
}
/// Retry the promise constructed in `body` up to `maxRetryCount` times.
///
/// - Note: Intentionally explicit about the recovery queue at the call site.
internal func attempt<T>(maxRetryCount: UInt, recoveringOn queue: DispatchQueue, body: @escaping () -> Promise<T>) -> Promise<T> {
var retryCount = 0
func attempt() -> Promise<T> {
return body().recover(on: queue) { error -> Promise<T> in
guard retryCount < maxRetryCount else { throw error }
retryCount += 1
return attempt()
}
return retryIfNeeded()
}
return attempt()
}

Write Preview
Loading…
Cancel
Save