From 8aa25af4d53066f1f0efcf0daac84490b68873f2 Mon Sep 17 00:00:00 2001 From: Niels Andriesse Date: Tue, 20 Apr 2021 14:58:29 +1000 Subject: [PATCH] Rework things in advance of new snode pool updating method --- SessionSnodeKit/Snode.swift | 1 + SessionSnodeKit/SnodeAPI.swift | 148 +++++++++++++++++++++------------ SessionSnodeKit/Storage.swift | 2 + 3 files changed, 97 insertions(+), 54 deletions(-) diff --git a/SessionSnodeKit/Snode.swift b/SessionSnodeKit/Snode.swift index ec6fbfce3..bc9a4eb8b 100644 --- a/SessionSnodeKit/Snode.swift +++ b/SessionSnodeKit/Snode.swift @@ -15,6 +15,7 @@ public final class Snode : NSObject, NSCoding { // NSObject/NSCoding conformance case getMessages = "retrieve" case getSessionIDForONSName = "get_lns_mapping" case sendMessage = "store" + case getAllSnodes = "get_all_service_nodes" } public struct KeySet { diff --git a/SessionSnodeKit/SnodeAPI.swift b/SessionSnodeKit/SnodeAPI.swift index b82d45ef7..ca8e4f85a 100644 --- a/SessionSnodeKit/SnodeAPI.swift +++ b/SessionSnodeKit/SnodeAPI.swift @@ -6,6 +6,7 @@ import Sodium public final class SnodeAPI : NSObject { private static var hasLoadedSnodePool = false private static var loadedSwarms: Set = [] + private static var getSnodePoolPromise: Promise>? /// - Note: Should only be accessed from `Threading.workQueue` to avoid race conditions. internal static var snodeFailureCount: [Snode:UInt] = [:] @@ -17,10 +18,11 @@ public final class SnodeAPI : NSObject { // MARK: Settings private static let maxRetryCount: UInt = 8 - private static let minimumSwarmSnodeCount = 3 + private static let minSwarmSnodeCount = 3 private static let seedNodePool: Set = useTestnet ? [ "http://public.loki.foundation:38157" ] : [ "https://storage.seed1.loki.network:4433", "https://storage.seed3.loki.network:4433", "https://public.loki.foundation:4433" ] private static let snodeFailureThreshold = 3 private static let targetSwarmSnodeCount = 2 + private static let minSnodePoolCount = 24; /// - Note: Changing this on the fly is not recommended. internal static var useOnionRequests = true @@ -61,14 +63,18 @@ public final class SnodeAPI : NSObject { hasLoadedSnodePool = true } - private static func setSnodePool(to newValue: Set, persist: Bool = true) { + private static func setSnodePool(to newValue: Set, using transaction: Any? = nil) { #if DEBUG dispatchPrecondition(condition: .onQueue(Threading.workQueue)) #endif snodePool = newValue - guard persist else { return } - SNSnodeKitConfiguration.shared.storage.writeSync { transaction in - SNSnodeKitConfiguration.shared.storage.setSnodePool(to: newValue, using: transaction) + let storage = SNSnodeKitConfiguration.shared.storage + if let transaction = transaction { + storage.setSnodePool(to: newValue, using: transaction) + } else { + storage.writeSync { transaction in + storage.setSnodePool(to: newValue, using: transaction) + } } } @@ -130,62 +136,96 @@ public final class SnodeAPI : NSObject { } internal static func getRandomSnode() -> Promise { + // randomElement() uses the system's default random generator, which is cryptographically secure + return getSnodePool().map2 { $0.randomElement()! } + } + + private static func getSnodePool() -> Promise> { + if let getSnodePoolPromise = getSnodePoolPromise { return getSnodePoolPromise } loadSnodePoolIfNeeded() let now = Date() - let isSnodePoolExpired = given(Storage.shared.getLastSnodePoolRefreshDate()) { now.timeIntervalSince($0) > 24 * 60 * 60 } ?? true - let isRefreshNeeded = (snodePool.isEmpty || isSnodePoolExpired) - if isRefreshNeeded { - SNSnodeKitConfiguration.shared.storage.write { transaction in - Storage.shared.setLastSnodePoolRefreshDate(to: now, using: transaction) - } - let target = seedNodePool.randomElement()! - let url = "\(target)/json_rpc" - let parameters: JSON = [ - "method" : "get_n_service_nodes", - "params" : [ - "active_only" : true, - "fields" : [ - "public_ip" : true, "storage_port" : true, "pubkey_ed25519" : true, "pubkey_x25519" : true - ] - ] - ] - SNLog("Populating snode pool using: \(target).") - let (promise, seal) = Promise.pending() - Threading.workQueue.async { - attempt(maxRetryCount: 4, recoveringOn: Threading.workQueue) { - HTTP.execute(.post, url, parameters: parameters, useSeedNodeURLSession: true).map2 { json -> Snode in - guard let intermediate = json["result"] as? JSON, let rawSnodes = intermediate["service_node_states"] as? [JSON] else { throw Error.snodePoolUpdatingFailed } - let snodePool: Set = Set(rawSnodes.compactMap { rawSnode in - guard let address = rawSnode["public_ip"] as? String, let port = rawSnode["storage_port"] as? Int, - let ed25519PublicKey = rawSnode["pubkey_ed25519"] as? String, let x25519PublicKey = rawSnode["pubkey_x25519"] as? String, address != "0.0.0.0" else { - SNLog("Failed to parse snode from: \(rawSnode).") - return nil - } - return Snode(address: "https://\(address)", port: UInt16(port), publicKeySet: Snode.KeySet(ed25519Key: ed25519PublicKey, x25519Key: x25519PublicKey)) - }) - setSnodePool(to: snodePool) - // randomElement() uses the system's default random generator, which is cryptographically secure - if !snodePool.isEmpty { - return snodePool.randomElement()! - } else { - throw Error.snodePoolUpdatingFailed - } - } - }.done2 { snode in - SNLog("Successfully updated snode pool.") - seal.fulfill(snode) - }.catch2 { error in - SNLog("Failed to contact seed node at: \(target).") - seal.reject(error) + let hasSnodePoolExpired = given(Storage.shared.getLastSnodePoolRefreshDate()) { now.timeIntervalSince($0) > 2 * 60 * 60 } ?? true + let snodePool = SnodeAPI.snodePool + let hasInsufficientSnodes = (snodePool.count < minSnodePoolCount) + if hasInsufficientSnodes || hasSnodePoolExpired { + let promise: Promise> + if snodePool.isEmpty { + promise = getSnodePoolFromSeedNode() + } else { + promise = getSnodePoolFromSnode().recover2 { _ in + getSnodePoolFromSeedNode() } } + getSnodePoolPromise = promise + promise.map2 { snodePool -> Set in + if snodePool.isEmpty { + throw Error.snodePoolUpdatingFailed + } else { + return snodePool + } + } + promise.then2 { snodePool -> Promise> in + let (promise, seal) = Promise>.pending() + SNSnodeKitConfiguration.shared.storage.write(with: { transaction in + Storage.shared.setLastSnodePoolRefreshDate(to: now, using: transaction) + setSnodePool(to: snodePool, using: transaction) + }, completion: { + seal.fulfill(snodePool) + }) + return promise + } + promise.done2 { _ in + getSnodePoolPromise = nil + } + promise.catch2 { _ in + getSnodePoolPromise = nil + } return promise } else { - return Promise { seal in - // randomElement() uses the system's default random generator, which is cryptographically secure - seal.fulfill(snodePool.randomElement()!) + return Promise.value(snodePool) + } + } + + private static func getSnodePoolFromSeedNode() -> Promise> { + let target = seedNodePool.randomElement()! + let url = "\(target)/json_rpc" + let parameters: JSON = [ + "method" : "get_n_service_nodes", + "params" : [ + "active_only" : true, + "fields" : [ + "public_ip" : true, "storage_port" : true, "pubkey_ed25519" : true, "pubkey_x25519" : true + ] + ] + ] + SNLog("Populating snode pool using seed node: \(target).") + let (promise, seal) = Promise>.pending() + Threading.workQueue.async { + attempt(maxRetryCount: 4, recoveringOn: Threading.workQueue) { + HTTP.execute(.post, url, parameters: parameters, useSeedNodeURLSession: true).map2 { json -> Set in + guard let intermediate = json["result"] as? JSON, let rawSnodes = intermediate["service_node_states"] as? [JSON] else { throw Error.snodePoolUpdatingFailed } + return Set(rawSnodes.compactMap { rawSnode in + guard let address = rawSnode["public_ip"] as? String, let port = rawSnode["storage_port"] as? Int, + let ed25519PublicKey = rawSnode["pubkey_ed25519"] as? String, let x25519PublicKey = rawSnode["pubkey_x25519"] as? String, address != "0.0.0.0" else { + SNLog("Failed to parse snode from: \(rawSnode).") + return nil + } + return Snode(address: "https://\(address)", port: UInt16(port), publicKeySet: Snode.KeySet(ed25519Key: ed25519PublicKey, x25519Key: x25519PublicKey)) + }) + } + }.done2 { snodePool in + SNLog("Got snode pool from seed node: \(target).") + seal.fulfill(snodePool) + }.catch2 { error in + SNLog("Failed to contact seed node at: \(target).") + seal.reject(error) } } + return promise + } + + private static func getSnodePoolFromSnode() -> Promise> { + return Promise(error: Error.generic) } // MARK: Public API @@ -261,7 +301,7 @@ public final class SnodeAPI : NSObject { public static func getSwarm(for publicKey: String) -> Promise> { loadSwarmIfNeeded(for: publicKey) - if let cachedSwarm = swarmCache[publicKey], cachedSwarm.count >= minimumSwarmSnodeCount { + if let cachedSwarm = swarmCache[publicKey], cachedSwarm.count >= minSwarmSnodeCount { return Promise> { $0.fulfill(cachedSwarm) } } else { SNLog("Getting swarm for: \((publicKey == SNSnodeKitConfiguration.shared.storage.getUserPublicKey()) ? "self" : publicKey).") diff --git a/SessionSnodeKit/Storage.swift b/SessionSnodeKit/Storage.swift index 878585618..f5b2050d4 100644 --- a/SessionSnodeKit/Storage.swift +++ b/SessionSnodeKit/Storage.swift @@ -5,6 +5,8 @@ public protocol SessionSnodeKitStorageProtocol { @discardableResult func write(with block: @escaping (Any) -> Void) -> Promise + @discardableResult + func write(with block: @escaping (Any) -> Void, completion: @escaping () -> Void) -> Promise func writeSync(with block: @escaping (Any) -> Void) func getUserPublicKey() -> String?