Improve loading speed on home timeline

This commit improves the loading speed for the home timeline (and likely
other areas of the app) by employing various techniques and changes:
- Network EOSE timeout reduced from 10 seconds down to 5 seconds
- Network EOSE does not wait on relays with broken connections
- Offload HomeModel handler event processing to separate tasks to
  avoid a large backlog
- Give SubscriptionManager streamers more fine-grained EOSE signals for
  local optimization
- Only wait for Ndb EOSE on the home timeline for faster loading
- Add logging with time elapsed measurements for easier identification of
  loading problems

Signed-off-by: Daniel D’Aquino <daniel@daquino.me>

damus/Core/Networking/NostrNetworkManager/SubscriptionManager.swift

@@ -18,6 +18,8 @@ extension NostrNetworkManager {
         private var taskManager: TaskManager
         private let experimentalLocalRelayModelSupport: Bool
         
+        let EXTRA_VERBOSE_LOGGING: Bool = false
+        
         init(pool: RelayPool, ndb: Ndb, experimentalLocalRelayModelSupport: Bool) {
             self.pool = pool
             self.ndb = ndb
@@ -28,17 +30,21 @@ extension NostrNetworkManager {
         // MARK: - Subscribing and Streaming data from Nostr
         
         /// Streams notes until the EOSE signal
-        func streamNotesUntilEndOfStoredEvents(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, timeout: Duration? = nil) -> AsyncStream<NdbNoteLender> {
+        func streamNotesUntilEndOfStoredEvents(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, timeout: Duration? = nil, id: UUID? = nil) -> AsyncStream<NdbNoteLender> {
             let timeout = timeout ?? .seconds(10)
             return AsyncStream<NdbNoteLender> { continuation in
                 let streamingTask = Task {
-                    outerLoop: for await item in self.subscribe(filters: filters, to: desiredRelays, timeout: timeout) {
+                    outerLoop: for await item in self.subscribe(filters: filters, to: desiredRelays, timeout: timeout, id: id) {
                         try Task.checkCancellation()
                         switch item {
                         case .event(let lender):
                             continuation.yield(lender)
                         case .eose:
                             break outerLoop
+                        case .ndbEose:
+                            continue
+                        case .networkEose:
+                            continue
                         }
                     }
                     continuation.finish()
@@ -52,10 +58,10 @@ extension NostrNetworkManager {
         /// Subscribes to data from user's relays, for a maximum period of time — after which the stream will end.
         ///
         /// This is useful when waiting for some specific data from Nostr, but not indefinitely.
-        func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, timeout: Duration) -> AsyncStream<StreamItem> {
+        func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, timeout: Duration, id: UUID? = nil) -> AsyncStream<StreamItem> {
             return AsyncStream<StreamItem> { continuation in
                 let streamingTask = Task {
-                    for await item in self.subscribe(filters: filters, to: desiredRelays) {
+                    for await item in self.subscribe(filters: filters, to: desiredRelays, id: id) {
                         try Task.checkCancellation()
                         continuation.yield(item)
                     }
@@ -79,9 +85,10 @@ extension NostrNetworkManager {
         ///
         /// - Parameter filters: The nostr filters to specify what kind of data to subscribe to
         /// - Returns: An async stream of nostr data
-        func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil) -> AsyncStream<StreamItem> {
+        func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, id: UUID? = nil) -> AsyncStream<StreamItem> {
             return AsyncStream<StreamItem> { continuation in
-                let subscriptionId = UUID()
+                let subscriptionId = id ?? UUID()
+                let startTime = CFAbsoluteTimeGetCurrent()
                 Log.info("Starting subscription %s: %s", for: .subscription_manager, subscriptionId.uuidString, filters.debugDescription)
                 let multiSessionStreamingTask = Task {
                     while !Task.isCancelled {
@@ -97,7 +104,7 @@ extension NostrNetworkManager {
                                 continue
                             }
                             Log.info("%s: Streaming.", for: .subscription_manager, subscriptionId.uuidString)
-                            for await item in self.sessionSubscribe(filters: filters, to: desiredRelays) {
+                            for await item in self.sessionSubscribe(filters: filters, to: desiredRelays, id: id) {
                                 try Task.checkCancellation()
                                 continuation.yield(item)
                             }
@@ -127,8 +134,11 @@ extension NostrNetworkManager {
         ///
         /// - Parameter filters: The nostr filters to specify what kind of data to subscribe to
         /// - Returns: An async stream of nostr data
-        private func sessionSubscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil) -> AsyncStream<StreamItem> {
+        private func sessionSubscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, id: UUID? = nil) -> AsyncStream<StreamItem> {
+            let id = id ?? UUID()
             return AsyncStream<StreamItem> { continuation in
+                let startTime = CFAbsoluteTimeGetCurrent()
+                Log.debug("Session subscription %s: Started", for: .subscription_manager, id.uuidString)
                 var ndbEOSEIssued = false
                 var networkEOSEIssued = false
                 
@@ -143,6 +153,7 @@ extension NostrNetworkManager {
                     (ndbEOSEIssued && (networkEOSEIssued || !connectedToNetwork))
                     
                     if canIssueEOSE {
+                        Log.debug("Session subscription %s: Issued EOSE for session. Elapsed: %.2f seconds", for: .subscription_manager, id.uuidString, CFAbsoluteTimeGetCurrent() - startTime)
                         continuation.yield(.eose)
                     }
                 }
@@ -153,7 +164,8 @@ extension NostrNetworkManager {
                             try Task.checkCancellation()
                             switch item {
                             case .eose:
-                                Log.debug("Session subscribe: Received EOSE from nostrdb", for: .subscription_manager)
+                                Log.debug("Session subscription %s: Received EOSE from nostrdb. Elapsed: %.2f seconds", for: .subscription_manager, id.uuidString, CFAbsoluteTimeGetCurrent() - startTime)
+                                continuation.yield(.ndbEose)
                                 ndbEOSEIssued = true
                                 yieldEOSEIfReady()
                             case .event(let noteKey):
@@ -170,32 +182,35 @@ extension NostrNetworkManager {
                         }
                     }
                     catch {
-                        Log.error("NDB streaming error: %s", for: .subscription_manager, error.localizedDescription)
+                        Log.error("Session subscription %s: NDB streaming error: %s", for: .subscription_manager, id.uuidString, error.localizedDescription)
                     }
                     continuation.finish()
                 }
                 let streamTask = Task {
                     do {
-                        for await item in self.pool.subscribe(filters: filters, to: desiredRelays) {
+                        for await item in self.pool.subscribe(filters: filters, to: desiredRelays, id: id) {
                             // NO-OP. Notes will be automatically ingested by NostrDB
                             // TODO: Improve efficiency of subscriptions?
                             try Task.checkCancellation()
                             switch item {
                             case .event(let event):
-                                Log.debug("Session subscribe: Received kind %d event with id %s from the network", for: .subscription_manager, event.kind, event.id.hex())
+                                if EXTRA_VERBOSE_LOGGING {
+                                    Log.debug("Session subscription %s: Received kind %d event with id %s from the network", for: .subscription_manager, id.uuidString, event.kind, event.id.hex())
+                                }
                                 if !self.experimentalLocalRelayModelSupport {
                                     // In normal mode (non-experimental), we stream from ndb but also directly from the network
                                     continuation.yield(.event(lender: NdbNoteLender(ownedNdbNote: event)))
                                 }
                             case .eose:
-                                Log.debug("Session subscribe: Received EOSE from the network", for: .subscription_manager)
+                                Log.debug("Session subscription %s: Received EOSE from the network. Elapsed: %.2f seconds", for: .subscription_manager, id.uuidString, CFAbsoluteTimeGetCurrent() - startTime)
+                                continuation.yield(.networkEose)
                                 networkEOSEIssued = true
                                 yieldEOSEIfReady()
                             }
                         }
                     }
                     catch {
-                        Log.error("Network streaming error: %s", for: .subscription_manager, error.localizedDescription)
+                        Log.error("Session subscription %s: Network streaming error: %s", for: .subscription_manager, id.uuidString, error.localizedDescription)
                     }
                     continuation.finish()
                 }
@@ -348,7 +363,27 @@ extension NostrNetworkManager {
     enum StreamItem {
         /// An event which can be borrowed from NostrDB
         case event(lender: NdbNoteLender)
-        /// The end of stored events
+        /// The canonical "end of stored events". See implementations of `subscribe` to see when this event is fired in relation to other EOSEs
         case eose
+        /// "End of stored events" from NostrDB.
+        case ndbEose
+        /// "End of stored events" from all relays in `RelayPool`.
+        case networkEose
+        
+        var debugDescription: String {
+            switch self {
+            case .event(lender: let lender):
+                let detailedDescription = try? lender.borrow({ event in
+                    "Note with ID: \(event.id.hex())"
+                })
+                return detailedDescription ?? "Some note"
+            case .eose:
+                return "EOSE"
+            case .ndbEose:
+                return "NDB EOSE"
+            case .networkEose:
+                return "NETWORK EOSE"
+            }
+        }
     }
 }

damus/Core/Networking/NostrNetworkManager/UserRelayListManager.swift

@@ -145,6 +145,8 @@ extension NostrNetworkManager {
                         try? self.set(userRelayList: relayList)                                         // Set the validated list
                     })
                 case .eose: continue
+                case .ndbEose: continue
+                case .networkEose: continue
                 }
             }
         }

damus/Core/Nostr/RelayPool.swift

@@ -232,11 +232,13 @@ class RelayPool {
     ///   - desiredRelays: The desired relays which to subsctibe to. If `nil`, it defaults to the `RelayPool`'s default list
     ///   - eoseTimeout: The maximum timeout which to give up waiting for the eoseSignal
     /// - Returns: Returns an async stream that callers can easily consume via a for-loop
-    func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, eoseTimeout: Duration? = nil) -> AsyncStream<StreamItem> {
+    func subscribe(filters: [NostrFilter], to desiredRelays: [RelayURL]? = nil, eoseTimeout: Duration? = nil, id: UUID? = nil) -> AsyncStream<StreamItem> {
-        let eoseTimeout = eoseTimeout ?? .seconds(10)
+        let eoseTimeout = eoseTimeout ?? .seconds(5)
-        let desiredRelays = desiredRelays ?? self.relays.map({ $0.descriptor.url })
+        let desiredRelays = desiredRelays ?? self.relays.filter({ $0.connection.isConnected }).map({ $0.descriptor.url })
+        let startTime = CFAbsoluteTimeGetCurrent()
         return AsyncStream<StreamItem> { continuation in
-            let sub_id = UUID().uuidString
+            let id = id ?? UUID()
+            let sub_id = id.uuidString
             var seenEvents: Set<NoteId> = []
             var relaysWhoFinishedInitialResults: Set<RelayURL> = []
             var eoseSent = false
@@ -257,6 +259,7 @@ class RelayPool {
                         break   // We do not support handling these yet
                     case .eose(_):
                         relaysWhoFinishedInitialResults.insert(relayUrl)
+                        Log.debug("RelayPool subscription %s: EOSE from %s. EOSE count: %d/%d. Elapsed: %.2f seconds.", for: .networking, id.uuidString, relayUrl.absoluteString, relaysWhoFinishedInitialResults.count, Set(desiredRelays).count, CFAbsoluteTimeGetCurrent() - startTime)
                         if relaysWhoFinishedInitialResults == Set(desiredRelays) {
                             continuation.yield(with: .success(.eose))
                             eoseSent = true

damus/Features/Chat/Models/ThreadModel.swift

@@ -122,6 +122,10 @@ class ThreadModel: ObservableObject {
                 case .eose:
                     guard let txn = NdbTxn(ndb: damus_state.ndb) else { return }
                     load_profiles(context: "thread", load: .from_events(Array(event_map.events)), damus_state: damus_state, txn: txn)
+                case .ndbEose:
+                    break
+                case .networkEose:
+                    break
                 }
             }
         }

damus/Features/Events/Models/EventsModel.swift

@@ -84,6 +84,10 @@ class EventsModel: ObservableObject {
                 case .eose:
                     DispatchQueue.main.async { self.loading = false }
                     break outerLoop
+                case .ndbEose:
+                    break
+                case .networkEose:
+                    break
                 }
             }
             DispatchQueue.main.async { self.loading = false }

damus/Features/FollowPack/Models/FollowPackModel.swift

@@ -59,6 +59,10 @@ class FollowPackModel: ObservableObject {
                 })
             case .eose:
                 continue
+            case .ndbEose:
+                continue
+            case .networkEose:
+                continue
             }
         }
     }

damus/Features/Follows/Models/FollowersModel.swift

@@ -45,6 +45,10 @@ class FollowersModel: ObservableObject {
                 case .eose:
                     guard let txn = NdbTxn(ndb: self.damus_state.ndb) else { return }
                     load_profiles(txn: txn)
+                case .ndbEose:
+                    continue
+                case .networkEose:
+                    continue
                 }
             }
         }
@@ -83,6 +87,10 @@ class FollowersModel: ObservableObject {
                 case .event(let lender):
                     lender.justUseACopy({ self.handle_event(ev: $0) })
                 case .eose: break
+                case .ndbEose:
+                    continue
+                case .networkEose:
+                    continue
                 }
             }
         }

damus/Features/NIP05/Models/NIP05DomainEventsModel.swift

@@ -73,6 +73,10 @@ class NIP05DomainEventsModel: ObservableObject {
                 load_profiles(context: "search", load: .from_events(self.events.all_events), damus_state: state, txn: txn)
                 DispatchQueue.main.async { self.loading = false }
                 continue
+            case .ndbEose:
+                break
+            case .networkEose:
+                break
             }
         }
     }

damus/Features/Onboarding/SuggestedUsersViewModel.swift

@@ -189,30 +189,25 @@ class SuggestedUsersViewModel: ObservableObject {
             authors: [Constants.ONBOARDING_FOLLOW_PACK_CURATOR_PUBKEY]
         )
         
-        for await item in self.damus_state.nostrNetwork.reader.subscribe(filters: [filter]) {
+        for await lender in self.damus_state.nostrNetwork.reader.streamNotesUntilEndOfStoredEvents(filters: [filter]) {
             // Check for cancellation on each iteration
             guard !Task.isCancelled else { break }
-
+            
-            switch item {
+            lender.justUseACopy({ event in
-            case .event(let lender):
+                let followPack = FollowPackEvent.parse(from: event)
-                lender.justUseACopy({ event in
+                
-                    let followPack = FollowPackEvent.parse(from: event)
+                guard let id = followPack.uuid else { return }
-                    
+                
-                    guard let id = followPack.uuid else { return }
+                let latestPackForThisId: FollowPackEvent
-                    
+                
-                    let latestPackForThisId: FollowPackEvent
+                if let existingPack = packsById[id], existingPack.event.created_at > followPack.event.created_at {
-                    
+                    latestPackForThisId = existingPack
-                    if let existingPack = packsById[id], existingPack.event.created_at > followPack.event.created_at {
+                } else {
-                        latestPackForThisId = existingPack
+                    latestPackForThisId = followPack
-                    } else {
+                }
-                        latestPackForThisId = followPack
+                
-                    }
+                packsById[id] = latestPackForThisId
-                    
+            })
-                    packsById[id] = latestPackForThisId
-                })
-            case .eose:
-                break
-            }
         }
     }
 
@@ -228,13 +223,8 @@ class SuggestedUsersViewModel: ObservableObject {
         }
         
         let profileFilter = NostrFilter(kinds: [.metadata], authors: allPubkeys)
-        for await item in damus_state.nostrNetwork.reader.subscribe(filters: [profileFilter]) {
+        for await _ in damus_state.nostrNetwork.reader.streamNotesUntilEndOfStoredEvents(filters: [profileFilter]) {
-            switch item {
+            // NO-OP. We just need NostrDB to ingest these for them to be available elsewhere, no need to analyze the data
-            case .event(_):
-                continue    // We just need NostrDB to ingest these for them to be available elsewhere, no need to analyze the data
-            case .eose:
-                break
-            }
         }
     }
 }

damus/Features/Profile/Models/ProfileModel.swift

@@ -81,6 +81,8 @@ class ProfileModel: ObservableObject, Equatable {
                 case .event(let lender):
                     lender.justUseACopy({ handleNostrEvent($0) })
                 case .eose: break
+                case .ndbEose: break
+                case .networkEose: break
                 }
             }
             guard let txn = NdbTxn(ndb: damus.ndb) else { return }
@@ -97,6 +99,8 @@ class ProfileModel: ObservableObject, Equatable {
                 case .event(let lender):
                     lender.justUseACopy({ handleNostrEvent($0) })
                 case .eose: break
+                case .ndbEose: break
+                case .networkEose: break
                 }
             }
             await bumpUpProgress()
@@ -138,6 +142,10 @@ class ProfileModel: ObservableObject, Equatable {
                 }
             case .eose:
                 continue
+            case .ndbEose:
+                continue
+            case .networkEose:
+                continue
             }
         }
     }
@@ -215,6 +223,10 @@ class ProfileModel: ObservableObject, Equatable {
                     }
                 case .eose:
                     break
+                case .ndbEose:
+                    break
+                case .networkEose:
+                    break
                 }
             }
         }

damus/Features/Timeline/Models/HomeModel.swift

@@ -450,6 +450,9 @@ class HomeModel: ContactsDelegate, ObservableObject {
     /// Send the initial filters, just our contact list and relay list mostly
     func send_initial_filters() {
         Task {
+            let startTime = CFAbsoluteTimeGetCurrent()
+            let id = UUID()
+            Log.info("Initial filter task started with ID %s", for: .homeModel, id.uuidString)
             let filter = NostrFilter(kinds: [.contacts], limit: 1, authors: [damus_state.pubkey])
             for await item in damus_state.nostrNetwork.reader.subscribe(filters: [filter]) {
                 switch item {
@@ -459,9 +462,14 @@ class HomeModel: ContactsDelegate, ObservableObject {
                 case .eose:
                     if !done_init {
                         done_init = true
+                        Log.info("Initial filter task %s: Done initialization; Elapsed time: %.2f seconds", for: .homeModel, id.uuidString, CFAbsoluteTimeGetCurrent() - startTime)
                         send_home_filters()
                     }
                     break
+                case .ndbEose:
+                    break
+                case .networkEose:
+                    break
                 }
             }
             
@@ -474,6 +482,8 @@ class HomeModel: ContactsDelegate, ObservableObject {
                 case .event(let lender):
                     await lender.justUseACopy({ await process_event(ev: $0, context: .initialRelayList) })
                 case .eose: break
+                case .ndbEose: break
+                case .networkEose: break
                 }
             }
         }
@@ -538,6 +548,8 @@ class HomeModel: ContactsDelegate, ObservableObject {
                 case .event(let lender):
                     await lender.justUseACopy({ await process_event(ev: $0, context: .contacts) })
                 case .eose: continue
+                case .ndbEose: continue
+                case .networkEose: continue
                 }
             }
         }
@@ -550,6 +562,8 @@ class HomeModel: ContactsDelegate, ObservableObject {
                 case .eose:
                     guard let txn = NdbTxn(ndb: damus_state.ndb) else { return }
                     load_profiles(context: "notifications", load: .from_keys(notifications.uniq_pubkeys()), damus_state: damus_state, txn: txn)
+                case .ndbEose: break
+                case .networkEose: break
                 }
             }
         }
@@ -564,6 +578,8 @@ class HomeModel: ContactsDelegate, ObservableObject {
                     var dms = dms.dms.flatMap { $0.events }
                     dms.append(contentsOf: incoming_dms)
                     load_profiles(context: "dms", load: .from_events(dms), damus_state: damus_state, txn: txn)
+                case .ndbEose: break
+                case .networkEose: break
                 }
             }
         }
@@ -580,6 +596,8 @@ class HomeModel: ContactsDelegate, ObservableObject {
                     case .event(let lender):
                         await lender.justUseACopy({ await process_event(ev: $0, context: .nwc) })
                     case .eose: continue
+                    case .ndbEose: continue
+                    case .networkEose: continue
                     }
                 }
             }
@@ -628,19 +646,43 @@ class HomeModel: ContactsDelegate, ObservableObject {
 
         self.homeHandlerTask?.cancel()
         self.homeHandlerTask = Task {
+            let startTime = CFAbsoluteTimeGetCurrent()
+            let id = UUID()
+            Log.info("Home handler task: Starting home handler task with ID %s", for: .homeModel, id.uuidString)
+    
             DispatchQueue.main.async {
                 self.loading = true
             }
-            for await item in damus_state.nostrNetwork.reader.subscribe(filters: home_filters) {
+            for await item in damus_state.nostrNetwork.reader.subscribe(filters: home_filters, id: id) {
                 switch item {
                 case .event(let lender):
-                    await lender.justUseACopy({ await process_event(ev: $0, context: .home) })
+                    let currentTime = CFAbsoluteTimeGetCurrent()
+                    // Process events in parallel on a separate task, to avoid holding up upcoming signals
+                    // Empirical evidence has shown that in at least one instance this technique saved up to 5 seconds of load time!
+                    Task { await lender.justUseACopy({ await process_event(ev: $0, context: .home) }) }
                 case .eose:
+                    let eoseTime = CFAbsoluteTimeGetCurrent()
+                    Log.info("Home handler task %s: Received general EOSE after %.2f seconds", for: .homeModel, id.uuidString, eoseTime - startTime)
+            
+                    guard let txn = NdbTxn(ndb: damus_state.ndb) else { return }
+                    load_profiles(context: "home", load: .from_events(events.events), damus_state: damus_state, txn: txn)
+            
+                    let finishTime = CFAbsoluteTimeGetCurrent()
+                    Log.info("Home handler task %s: Completed initial loading task after %.2f seconds", for: .homeModel, id.uuidString, eoseTime - startTime)
+                case .ndbEose:
+                    let eoseTime = CFAbsoluteTimeGetCurrent()
+                    Log.info("Home handler task %s: Received NDB EOSE after %.2f seconds", for: .homeModel, id.uuidString, eoseTime - startTime)
+            
                     guard let txn = NdbTxn(ndb: damus_state.ndb) else { return }
                     DispatchQueue.main.async {
                         self.loading = false
                     }
                     load_profiles(context: "home", load: .from_events(events.events), damus_state: damus_state, txn: txn)
+            
+                    let finishTime = CFAbsoluteTimeGetCurrent()
+                    Log.info("Home handler task %s: Completed initial NDB loading task after %.2f seconds", for: .homeModel, id.uuidString, eoseTime - startTime)
+                case .networkEose:
+                    break
                 }
             }
         }

damus/Features/Wallet/Models/WalletModel.swift

@@ -198,6 +198,10 @@ class WalletModel: ObservableObject {
                 return result
             case .eose:
                 continue
+            case .ndbEose:
+                continue
+            case .networkEose:
+                continue
             }
         }
         do { try Task.checkCancellation() } catch { throw .cancelled }

damus/Features/Zaps/Models/ZapsModel.swift

@@ -43,6 +43,10 @@ class ZapsModel: ObservableObject {
                     let events = state.events.lookup_zaps(target: target).map { $0.request.ev }
                     guard let txn = NdbTxn(ndb: state.ndb) else { return }
                     load_profiles(context: "zaps_model", load: .from_events(events), damus_state: state, txn: txn)
+                case .ndbEose:
+                    break
+                case .networkEose:
+                    break
                 }
             }
         }

damus/Shared/Utilities/Log.swift

@@ -24,6 +24,7 @@ enum LogCategory: String {
     case video_coordination
     case tips
     case ndb
+    case homeModel
 }
 
 /// Damus structured logger

damusTests/NostrNetworkManagerTests/NostrNetworkManagerTests.swift

@@ -61,6 +61,10 @@ class NostrNetworkManagerTests: XCTestCase {
                 case .eose:
                     // End of stream, break out of the loop
                     endOfStream.fulfill()
+                case .ndbEose:
+                    continue
+                case .networkEose:
+                    continue
                 }
             }
         }