feat(sync): re-arm the open channel's subscription first on reconnect

App/Sources/Timeline/ChannelTimelineView.swift

@@ -261,6 +261,12 @@ struct ChannelTimelineView: View {
         // the replies themselves, so on a cold launch a row would otherwise say "3 replies"
         // over an empty strip. See ``BuzzKit/SyncEngine/prefetchThreads(in:)``.
         .task { await prefetcher?.prefetchThreads(in: channelID) }
+        // Which conversation is on screen, so that a reconnect — which is most of what
+        // foregrounding the app does — restores *this* channel's live subscription
+        // before the other joined channels' (§ ``BuzzKit/SyncEngine/setActiveChannel(_:)``).
+        // Not cleared when the view goes away: the engine's preference is advisory, and
+        // the channel just left is the one most likely to be opened again.
+        .task { await lifecycleEngine?.setActiveChannel(channelID) }
     }
 
     /// How this conversation presents itself — a channel, or the person on the other

Packages/BuzzKit/Sources/BuzzKit/SyncEngine+ChannelSubscriptions.swift

@@ -133,9 +133,37 @@ extension SyncEngine {
             return winner
         }
         channelContentSubscriptions[channel] = id
+        // A conversation can be on screen before discovery has registered its channel —
+        // opening one is itself a route into this method. Claim the re-arm priority now
+        // rather than waiting for the next ``setActiveChannel(_:)``, which for a channel
+        // already open would never come.
+        if channel == activeChannel {
+            await subscriptions.prioritise(id)
+        }
         return id
     }
 
+    // MARK: - Re-arm priority
+
+    /// Reports which channel the reader has on screen, so a reconnect restores that
+    /// channel's live traffic before every other channel's.
+    ///
+    /// A reconnect re-`REQ`s every standing subscription and the engine keeps one per
+    /// joined channel, so without a stated preference the open conversation took its
+    /// turn in an arbitrary order — on a busy account, potentially last. This names it,
+    /// and the ``SubscriptionManager`` arms it first.
+    ///
+    /// Ordering only: membership, filters and delivery are all untouched, and a value
+    /// that is stale or names a channel with no subscription is skipped. So it is safe
+    /// to call before the channel is subscribed — ``subscribeChannelContent(_:)`` picks
+    /// the priority up when it registers one — and there is no obligation to clear it
+    /// when the conversation closes. Leaving the last-read channel prioritised is the
+    /// better resting state anyway: it is where the reader most likely returns.
+    public func setActiveChannel(_ channel: String?) async {
+        activeChannel = channel
+        await subscriptions.prioritise(channel.flatMap { channelContentSubscriptions[$0] })
+    }
+
     /// Drops the standing content subscription for `channel` with a `CLOSE`. A no-op
     /// when none is open.
     func unsubscribeChannelContent(_ channel: String) async {

Packages/BuzzKit/Sources/BuzzKit/SyncEngine.swift

@@ -238,6 +238,15 @@ public actor SyncEngine {
     /// only in ``stop()``.
     var channelContentSubscriptions: [String: SubscriptionID] = [:]
 
+    /// The channel the reader has on screen, as last reported by the app through
+    /// ``setActiveChannel(_:)``.
+    ///
+    /// Held rather than just forwarded so that a channel opened *before* its standing
+    /// subscription exists still becomes the re-arm priority the moment
+    /// ``subscribeChannelContent(_:)`` registers one. Ordering only — see
+    /// ``setActiveChannel(_:)``.
+    var activeChannel: String?
+
     // MARK: - Tasks
 
     private var stateObserverTask: Task<Void, Never>?
@@ -454,6 +463,9 @@ public actor SyncEngine {
         // content-subscription ids so a later reopen re-registers cleanly rather than
         // believing a channel is still subscribed.
         channelContentSubscriptions.removeAll()
+        // Named an id in the table just cleared; a later session re-reports whatever
+        // is on screen then.
+        activeChannel = nil
         state = .stopped
     }
 

Packages/NostrCore/Sources/NostrCore/SubscriptionManager+Subscribing.swift

@@ -18,11 +18,26 @@ extension SubscriptionManager {
         let epoch = readyEpoch
         // Snapshot the ids: `armSubscription` awaits a send, during which a
         // reentrant register or unsubscribe may mutate the table.
-        for id in Array(subscriptions.keys) {
+        for id in armOrder() {
             await armSubscription(id, epoch: epoch, resetCloseRetry: true)
         }
     }
 
+    /// Every registered subscription id, with ``SubscriptionManager/prioritySubscriptionID``
+    /// first when it still names a live one.
+    ///
+    /// The rest keep `Dictionary`'s order. That order is arbitrary, and for them it is
+    /// also of no consequence — what mattered was that the subscription a reader is
+    /// watching was somewhere in it, taking its chances against every other channel's
+    /// re-`REQ`.
+    private func armOrder() -> [SubscriptionID] {
+        let ids = Array(subscriptions.keys)
+        guard let priority = prioritySubscriptionID, subscriptions[priority] != nil else {
+            return ids
+        }
+        return [priority] + ids.filter { $0 != priority }
+    }
+
     /// Sends the `REQ` for a subscription under a readiness epoch, at most once
     /// per epoch. A repeat call for an epoch already served is a no-op, so
     /// registration and the readiness observer cannot double-`REQ` the same

Packages/NostrCore/Sources/NostrCore/SubscriptionManager.swift

@@ -103,6 +103,18 @@ public actor SubscriptionManager {
     /// every live subscription must be re-`REQ`ed onto exactly once.
     var readyEpoch = 0
 
+    /// The subscription to re-`REQ` first when the connection returns to `ready`.
+    ///
+    /// Re-arming walks every live subscription, and until this existed it walked them
+    /// in `Dictionary`'s order — so the conversation a reader is actually looking at
+    /// could be restored last, and its channel stayed silent for the whole replay. A
+    /// consumer that knows which subscription is on screen names it here.
+    ///
+    /// Advisory, never load-bearing. A stale id — its subscription unsubscribed, or
+    /// closed by the relay — is skipped and the remaining order is unchanged, so no
+    /// caller has to clear it to keep re-arming correct.
+    var prioritySubscriptionID: SubscriptionID?
+
     /// The authenticated identity's hex pubkey, resolved from the signer once and
     /// cached — only pubkey-gated filters need it, and only the first time.
     var cachedPubkeyHex: String?
@@ -164,6 +176,19 @@ public actor SubscriptionManager {
         subscriptions.removeAll()
         connectionIsReady = false
         started = false
+        prioritySubscriptionID = nil
+    }
+
+    // MARK: - Re-arm priority
+
+    /// Names the subscription to re-`REQ` first on the next return to `ready` — the
+    /// one whose events a reader is waiting on. `nil` drops the preference.
+    ///
+    /// Ordering only. It arms nothing by itself, and it changes neither what a
+    /// subscription asks for nor what it receives, so naming a subscription that is
+    /// gone, or naming none, costs only the ordering it asked for.
+    public func prioritise(_ id: SubscriptionID?) {
+        prioritySubscriptionID = id
     }
 
     // MARK: - Registration

Packages/NostrCore/Tests/NostrCoreTests/SubscriptionManagerTests.swift

@@ -224,6 +224,70 @@ struct SubscriptionManagerTests {
         await connection.stop()
     }
 
+    @Test("The prioritised subscription is re-REQed before the others on reconnect", .timeLimit(.minutes(1)))
+    func prioritisedSubscriptionReArmsFirst() async throws {
+        let signer = try InMemorySigner()
+        let first = FakeRelay()
+        let second = FakeRelay()
+        let transports = TransportQueue([first, second])
+        let connection = makeInertConnection(signer: signer, transports: transports)
+        let manager = SubscriptionManager(connection: connection, signer: signer)
+
+        try await connection.connect()
+        try await driveAuthToReady(connection, first, authSendIndex: 0)
+
+        // Three subscriptions, so the assertion cannot pass on a coin flip: only one of
+        // the six possible orders puts the *last* registered one first.
+        var ids: [SubscriptionID] = []
+        for kind in [EventKind.channelMessage, .richMessage, .reaction] {
+            ids.append(try await manager.register(filters: [Filter(kinds: [kind])], sink: RecordingSink()))
+            _ = await first.awaitSend(index: ids.count) // initial REQ on the first socket
+        }
+        let last = try #require(ids.last)
+        await manager.prioritise(last)
+
+        await first.enqueueFailure(.connectionClosed)
+        try await driveAuthToReady(connection, second, authSendIndex: 0)
+
+        let firstReArmed = await second.awaitSend(index: 1)
+        #expect(try reqSubscriptionID(from: firstReArmed) == last.rawValue)
+
+        await manager.shutdown()
+        await connection.stop()
+    }
+
+    @Test("A priority naming a subscription that is gone leaves re-arming intact", .timeLimit(.minutes(1)))
+    func stalePriorityDoesNotStrandReArming() async throws {
+        let signer = try InMemorySigner()
+        let first = FakeRelay()
+        let second = FakeRelay()
+        let transports = TransportQueue([first, second])
+        let connection = makeInertConnection(signer: signer, transports: transports)
+        let manager = SubscriptionManager(connection: connection, signer: signer)
+
+        try await connection.connect()
+        try await driveAuthToReady(connection, first, authSendIndex: 0)
+        let survivor = try await manager.register(filters: [Filter(kinds: [.channelMessage])], sink: RecordingSink())
+        _ = await first.awaitSend(index: 1)
+        let doomed = try await manager.register(filters: [Filter(kinds: [.richMessage])], sink: RecordingSink())
+        _ = await first.awaitSend(index: 2)
+
+        // Prioritise, then unsubscribe the very subscription named. The stale id must be
+        // skipped rather than consume the first slot or abort the walk.
+        await manager.prioritise(doomed)
+        await manager.unsubscribe(doomed)
+        _ = await first.awaitSend(index: 3) // CLOSE
+
+        await first.enqueueFailure(.connectionClosed)
+        try await driveAuthToReady(connection, second, authSendIndex: 0)
+
+        let reArmed = await second.awaitSend(index: 1)
+        #expect(try reqSubscriptionID(from: reArmed) == survivor.rawValue)
+
+        await manager.shutdown()
+        await connection.stop()
+    }
+
     // MARK: - Unsubscribe
 
     @Test("Unsubscribe sends CLOSE and drops later frames for the dead subscription", .timeLimit(.minutes(1)))