From 75a10fbbd0b203cb598b85bc949e7a7790f84c5d Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Fri, 2 Oct 2026 15:32:35 +0800 Subject: [PATCH] feat(pro): gate multi-connection acceleration without changing saved settings --- .../MultiConnectionSettingsTests.swift | 75 +++++++++++++ FrameworkTests/RevenueCatCommerceTests.swift | 17 +++ Frameworks/Resources/Localizable.xcstrings | 51 +++++++++ .../Commerce/ProAccessTypes.swift | 2 + .../Commerce/ProAccessController.swift | 15 +++ .../MultiConnectionSettingsAccess.swift | 38 +++++++ .../SFTP/SFTPSettingsFields.swift | 41 +++---- .../TermindPresentation/Strings/Pro.swift | 4 + .../TermindPresentation/Strings/SFTP.swift | 12 +++ .../SSH/PortForwardingController.swift | 44 ++++++-- .../SSH/SSHForwardingConnection.swift | 11 ++ .../SSH/SSHForwardingConnectionPool.swift | 101 ++++++++++++++++-- .../Transfers/SFTPTransferCoordinator.swift | 7 +- TFI/App/MobileAppDependencies.swift | 1 + TFI/App/MobileAppModel.swift | 20 ++-- .../Settings/MobileSFTPSettingsView.swift | 3 +- .../Settings/MobileSettingsView.swift | 6 +- .../MobileMultiConnectionAccessTests.swift | 37 +++++++ TFM/App/AppModel.swift | 22 ++-- TFM/Features/HostEditor/HostEditorView.swift | 14 ++- TFM/Features/Settings/SettingsPageView.swift | 8 +- TFMTests/PortForwardingReconnectTests.swift | 35 +++++- TFMTests/SFTPConcurrencySchedulerTests.swift | 24 +++++ TFMTests/SFTPSettingsIntegrationTests.swift | 83 ++++++++++++++ TFMTests/SSHForwardingWireTests.swift | 70 ++++++++++++ 25 files changed, 678 insertions(+), 63 deletions(-) create mode 100644 FrameworkTests/MultiConnectionSettingsTests.swift create mode 100644 Frameworks/TermindPresentation/MultiConnectionSettingsAccess.swift create mode 100644 TFITests/MobileMultiConnectionAccessTests.swift diff --git a/FrameworkTests/MultiConnectionSettingsTests.swift b/FrameworkTests/MultiConnectionSettingsTests.swift new file mode 100644 index 00000000..555dc8ce --- /dev/null +++ b/FrameworkTests/MultiConnectionSettingsTests.swift @@ -0,0 +1,75 @@ +import Foundation +import SwiftUI +import TermindApplication +import TermindDomain +import TermindInfrastructure +import TermindPresentation +import Testing + +@Suite @MainActor +struct MultiConnectionSettingsTests { + @Test(arguments: SFTPConcurrencyStrategy.allCases) + func freeDisplayAndOtherEditsPreserveTheSavedStrategy(_ strategy: SFTPConcurrencyStrategy) { + let persistence = SettingsStore(.init(strategy: strategy)) + let settings = SFTPSettingsModel(persistence: persistence) + let pro = ProAccessController.testing(status: .free(trialAvailable: true)) + let access = MultiConnectionSettingsAccess(isAllowed: false) { pro.require(.sftpMultiConnection) } + let selection = access.strategy(for: settings) + #expect(selection.wrappedValue == .channelsOnly) + #expect(persistence.saves == 0) + selection.wrappedValue = .channelsOnly + #expect(persistence.saves == 0) + selection.wrappedValue = .smart + selection.wrappedValue = .connectionsPreferred + #expect(pro.isPaywallPresented) + #expect(persistence.saves == 0) + #expect(settings.set(12, for: \.globalLimit)) + #expect(persistence.value.strategy == strategy) + #expect(persistence.saves == 1) + let upgraded = MultiConnectionSettingsAccess(isAllowed: true, requestAccess: {}) + #expect(upgraded.strategy(for: settings).wrappedValue == strategy) + } + + @Test func freeHostEditorSavesOtherFieldsWithoutReplacingItsConnectionCount() { + var host = Host(name: "Before", host: "example.invalid", portForwardingMaxConnections: 8) + var prompts = 0 + let saved = Binding(get: { host.portForwardingMaxConnections }, set: { host.portForwardingMaxConnections = $0 }) + let free = MultiConnectionSettingsAccess(isAllowed: false) { prompts += 1 }.forwardingConnections(saved) + #expect(free.wrappedValue == 1) + free.wrappedValue = 1 + free.wrappedValue = 2 + host.name = "After" + #expect(host.portForwardingMaxConnections == 8) + #expect(prompts == 1) + let paid = MultiConnectionSettingsAccess(isAllowed: true, requestAccess: {}).forwardingConnections(saved) + #expect(paid.wrappedValue == 8) + paid.wrappedValue = 4 + #expect(host.portForwardingMaxConnections == 4) + } + + @Test(arguments: SFTPConcurrencyStrategy.allCases) + func runtimeProjectionPreservesSavedPreferencesWithoutOpeningPaywall(_ strategy: SFTPConcurrencyStrategy) { + let saved = SFTPPreferences(globalLimit: 80, perHostLimit: 40, strategy: strategy) + let free = ProAccessController.testing(status: .verificationFailed) + let effective = free.effectiveSFTPPreferences(saved) + #expect(effective == SFTPPreferences(globalLimit: 80, perHostLimit: 40, strategy: .channelsOnly)) + #expect(free.effectiveForwardingConnections(8) == 1) + #expect(!free.isPaywallPresented) + let paid = ProAccessController.testing(status: .purchased(.familyShared)) + #expect(paid.effectiveSFTPPreferences(saved) == saved) + #expect(paid.effectiveForwardingConnections(8) == 8) + #expect(!paid.isPaywallPresented) + } +} + +@MainActor +private final class SettingsStore: SFTPPreferencesPersisting { + var value: SFTPPreferences + var saves = 0 + init(_ value: SFTPPreferences) { self.value = value } + func loadSFTP() throws -> SFTPPreferences { value } + func saveSFTP(_ value: SFTPPreferences) throws { + self.value = value + saves += 1 + } +} diff --git a/FrameworkTests/RevenueCatCommerceTests.swift b/FrameworkTests/RevenueCatCommerceTests.swift index 020dc502..071ce6b7 100644 --- a/FrameworkTests/RevenueCatCommerceTests.swift +++ b/FrameworkTests/RevenueCatCommerceTests.swift @@ -116,6 +116,23 @@ struct RevenueCatCommerceTests { #expect(!controller.hasProAccess) } + @Test func accessChangesNotifyRuntimeWithoutAVisibleViewOrPaywall() async { + let service = FakeService() + let controller = ProAccessController(purchaseService: service, observeUpdates: false) + var observed: [Bool] = [] + controller.accessDidChange = { observed.append(controller.hasProAccess) } + defer { controller.accessDidChange = nil } + await controller.refresh() + service.status = .purchased(.purchased) + await controller.refresh() + service.status = .purchased(.familyShared) + await controller.refresh() + service.status = .free(trialAvailable: false) + await controller.refresh() + #expect(observed == [true, false]) + #expect(!controller.isPaywallPresented) + } + @Test func newerCustomerInfoWinsOverAnOlderRefreshResponse() async { let service = FakeService() diff --git a/Frameworks/Resources/Localizable.xcstrings b/Frameworks/Resources/Localizable.xcstrings index 04fd9ddf..45abab39 100644 --- a/Frameworks/Resources/Localizable.xcstrings +++ b/Frameworks/Resources/Localizable.xcstrings @@ -7223,6 +7223,23 @@ } } }, + "Connections Preferred (Pro)" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "优先连接（Pro）" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "優先連線（Pro）" + } + } + } + }, "Contacting the App Store…" : { "extractionState" : "manual", "localizations" : { @@ -15919,6 +15936,23 @@ } } }, + "Max Connections (Pro)" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "最大连接数（Pro）" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "連線數上限（Pro）" + } + } + } + }, "Max Output" : { "extractionState" : "manual", "localizations" : { @@ -26097,6 +26131,23 @@ } } }, + "Smart (Pro)" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "智能（Pro）" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "智慧（Pro）" + } + } + } + }, "Smart uses channels for files up to 64 MB and prefers additional connections for larger files. Transfers with an unknown size use channels." : { "extractionState" : "manual", "localizations" : { diff --git a/Frameworks/TermindDomain/Commerce/ProAccessTypes.swift b/Frameworks/TermindDomain/Commerce/ProAccessTypes.swift index d85e723c..98ff71a5 100644 --- a/Frameworks/TermindDomain/Commerce/ProAccessTypes.swift +++ b/Frameworks/TermindDomain/Commerce/ProAccessTypes.swift @@ -6,6 +6,8 @@ public nonisolated enum ProFeature: String, CaseIterable, Hashable, Sendable { case mcpServer case vaultSync case unlimitedMetrics + case sftpMultiConnection + case portForwardingMultiConnection } public nonisolated enum ProOwnership: Equatable, Sendable { diff --git a/Frameworks/TermindInfrastructure/Commerce/ProAccessController.swift b/Frameworks/TermindInfrastructure/Commerce/ProAccessController.swift index 5d76e5bd..3b1be487 100644 --- a/Frameworks/TermindInfrastructure/Commerce/ProAccessController.swift +++ b/Frameworks/TermindInfrastructure/Commerce/ProAccessController.swift @@ -15,6 +15,8 @@ public final class ProAccessController { public private(set) var status: EntitlementStatus public private(set) var isProcessing = false public var isPaywallPresented = false + /// App-owned runtime reconciliation; does not require a visible SwiftUI surface. + public var accessDidChange: (() -> Void)? public var usesRevenueCat: Bool { purchaseService != nil } public var hasProAccess: Bool { status.hasProAccess } @@ -62,6 +64,17 @@ public final class ProAccessController { public func canUse(_: ProFeature) -> Bool { hasProAccess } + /// Resolve access without changing the user's persisted choice or presenting commerce UI. + public func effectiveSFTPPreferences(_ saved: SFTPPreferences) -> SFTPPreferences { + var effective = saved + if !canUse(.sftpMultiConnection) { effective.strategy = .channelsOnly } + return effective + } + + public func effectiveForwardingConnections(_ saved: Int) -> Int { + canUse(.portForwardingMultiConnection) ? Host.normalizedPortForwardingMaxConnections(saved) : 1 + } + @discardableResult public func require(_ feature: ProFeature) -> Bool { guard !canUse(feature) else { return true } @@ -118,7 +131,9 @@ public final class ProAccessController { } private func apply(_ next: EntitlementStatus) { + let changed = status.hasProAccess != next.hasProAccess status = next + if changed { accessDidChange?() } entitlementExpirationTask?.cancel() if case .subscription(let subscription) = next, subscription.phase.hasProAccess, let expiresAt = subscription.expiresAt, expiresAt > now() diff --git a/Frameworks/TermindPresentation/MultiConnectionSettingsAccess.swift b/Frameworks/TermindPresentation/MultiConnectionSettingsAccess.swift new file mode 100644 index 00000000..8ff2918f --- /dev/null +++ b/Frameworks/TermindPresentation/MultiConnectionSettingsAccess.swift @@ -0,0 +1,38 @@ +import SwiftUI +import TermindApplication +import TermindDomain + +/// Editing-only projections. Reading a free value never overwrites the saved choice. +public struct MultiConnectionSettingsAccess { + public var isAllowed: Bool + private let requestAccess: () -> Void + + public init(isAllowed: Bool, requestAccess: @escaping () -> Void) { + self.isAllowed = isAllowed + self.requestAccess = requestAccess + } + + public func strategy(for model: SFTPSettingsModel) -> Binding { + Binding( + get: { isAllowed ? model.preferences.strategy : .channelsOnly }, + set: { value in + guard isAllowed else { + if value != .channelsOnly { requestAccess() } + return + } + model.setStrategy(value) + }) + } + + public func forwardingConnections(_ saved: Binding) -> Binding { + Binding( + get: { isAllowed ? saved.wrappedValue : 1 }, + set: { value in + guard isAllowed else { + if value > 1 { requestAccess() } + return + } + saved.wrappedValue = Host.normalizedPortForwardingMaxConnections(value) + }) + } +} diff --git a/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift b/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift index 54292cca..caa5f8d2 100644 --- a/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift +++ b/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift @@ -4,31 +4,35 @@ import TermindDomain public struct SFTPSettingsFields: View { let model: SFTPSettingsModel + let access: MultiConnectionSettingsAccess - public init(model: SFTPSettingsModel) { self.model = model } - + public init( + model: SFTPSettingsModel, + access: MultiConnectionSettingsAccess = .init(isAllowed: true, requestAccess: {}) + ) { + self.model = model + self.access = access + } public var body: some View { - Picker("Concurrent Strategy", selection: strategyBinding) { - Text("Smart").tag(SFTPConcurrencyStrategy.smart) - Text("Channels Only").tag(SFTPConcurrencyStrategy.channelsOnly) - Text("Connections Preferred").tag(SFTPConcurrencyStrategy.connectionsPreferred) + Picker(AppStrings.SFTP.concurrentStrategy, selection: access.strategy(for: model)) { + Text(access.isAllowed ? AppStrings.SFTP.smart : AppStrings.Pro.smartStrategy) + .tag(SFTPConcurrencyStrategy.smart) + Text(AppStrings.SFTP.channelsOnly).tag(SFTPConcurrencyStrategy.channelsOnly) + Text(access.isAllowed ? AppStrings.SFTP.connectionsPreferred : AppStrings.Pro.connectionsPreferredStrategy) + .tag(SFTPConcurrencyStrategy.connectionsPreferred) } - if model.preferences.strategy == .smart { - Text( - "Smart uses channels for files up to 64 MB and prefers additional connections for larger files. Transfers with an unknown size use channels." - ) - .font(.caption).foregroundStyle(.secondary) + if access.strategy(for: model).wrappedValue == .smart { + Text(AppStrings.SFTP.smartStrategyExplanation) + .font(.caption).foregroundStyle(.secondary) } Stepper(value: binding(\.globalLimit), in: SFTPPreferences.globalLimitRange) { - label("Concurrent Transfers", value: model.preferences.globalLimit) + label(AppStrings.SFTP.concurrentTransfers, value: model.preferences.globalLimit) } Stepper(value: binding(\.perHostLimit), in: model.preferences.availablePerHostRange) { - label("Per-Host Transfers", value: model.preferences.perHostLimit) + label(AppStrings.SFTP.perHostTransfers, value: model.preferences.perHostLimit) } - Text( - "Changes apply to queued transfers; active transfers continue. If new connections would prompt for authentication, transfers use parallel channels on the existing connection." - ) - .font(.caption).foregroundStyle(.secondary) + Text(AppStrings.SFTP.concurrencyChangesExplanation) + .font(.caption).foregroundStyle(.secondary) if let error = model.storageError { Text(error).foregroundStyle(.red) } } @@ -51,7 +55,4 @@ public struct SFTPSettingsFields: View { Binding(get: { model.preferences[keyPath: keyPath] }, set: { model.set($0, for: keyPath) }) } - private var strategyBinding: Binding { - Binding(get: { model.preferences.strategy }, set: { model.setStrategy($0) }) - } } diff --git a/Frameworks/TermindPresentation/Strings/Pro.swift b/Frameworks/TermindPresentation/Strings/Pro.swift index 5e92ee08..848597fe 100644 --- a/Frameworks/TermindPresentation/Strings/Pro.swift +++ b/Frameworks/TermindPresentation/Strings/Pro.swift @@ -2,6 +2,10 @@ import Foundation extension AppStrings { public nonisolated enum Pro { + public static var smartStrategy: LocalizedStringResource { "Smart (Pro)" } + public static var connectionsPreferredStrategy: LocalizedStringResource { "Connections Preferred (Pro)" } + public static var maxConnections: LocalizedStringResource { "Max Connections (Pro)" } + public static var contactingTheAppStoreProgress: LocalizedStringResource { "Contacting the App Store…" } diff --git a/Frameworks/TermindPresentation/Strings/SFTP.swift b/Frameworks/TermindPresentation/Strings/SFTP.swift index 01a8a601..233b60a6 100644 --- a/Frameworks/TermindPresentation/Strings/SFTP.swift +++ b/Frameworks/TermindPresentation/Strings/SFTP.swift @@ -2,6 +2,18 @@ import Foundation extension AppStrings { public nonisolated enum SFTP { + public static var concurrentStrategy: LocalizedStringResource { "Concurrent Strategy" } + public static var smart: LocalizedStringResource { "Smart" } + public static var channelsOnly: LocalizedStringResource { "Channels Only" } + public static var connectionsPreferred: LocalizedStringResource { "Connections Preferred" } + public static var concurrentTransfers: LocalizedStringResource { "Concurrent Transfers" } + public static var perHostTransfers: LocalizedStringResource { "Per-Host Transfers" } + public static var smartStrategyExplanation: LocalizedStringResource { + "Smart uses channels for files up to 64 MB and prefers additional connections for larger files. Transfers with an unknown size use channels." + } + public static var concurrencyChangesExplanation: LocalizedStringResource { + "Changes apply to queued transfers; active transfers continue. If new connections would prompt for authentication, transfers use parallel channels on the existing connection." + } public static var appStoreProgress: LocalizedStringResource { "App Store…" diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift b/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift index 13b0c016..40903543 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift @@ -80,7 +80,7 @@ public final class PortForwardingConnection { fileprivate var silentAttempt = false fileprivate var additionalSnapshot: SSHAdditionalConnectionSnapshot? fileprivate var connectionCounts: [PortForwardingRule.ID: Int] = [:] - fileprivate let maximumConnections: Int + fileprivate var maximumConnections: Int fileprivate init( host: Host, route: [Host], maximumConnections: Int, reconnect: SessionReconnectController @@ -100,14 +100,6 @@ public final class PortForwardingController { public nonisolated enum PoolPolicy: Sendable { case singleConnection case hostConfigured - - fileprivate func maximumConnections(for host: Host) -> Int { - switch self { - case .singleConnection: 1 - case .hostConfigured: - Host.normalizedPortForwardingMaxConnections(host.portForwardingMaxConnections) - } - } } public typealias ConnectionFactory = @@ -122,6 +114,8 @@ public final class PortForwardingController { @escaping SSHForwardingConnection.FailureHandler ) -> any SSHForwardingConnectionProtocol private let poolPolicy: PoolPolicy + private var connectionLimit: Int + private var limitUpdateTask: Task? private let logger: Logger private let makeReconnectController: @MainActor () -> SessionReconnectController public private(set) var states: [PortForwardingRule.ID: PortForwardingRuntimeState] = [:] @@ -135,6 +129,7 @@ public final class PortForwardingController { public init( hosts: HostLibraryModel, poolPolicy: PoolPolicy = .singleConnection, + connectionLimit: Int = Host.portForwardingMaxConnectionsRange.upperBound, logger: Logger = ApplicationLogger.disabled("termind.port-forwarding"), makeConnection: ConnectionFactory? = nil, makeReconnectController: @escaping @MainActor () -> SessionReconnectController = { @@ -144,6 +139,7 @@ public final class PortForwardingController { self.hosts = hosts self.logger = logger self.poolPolicy = poolPolicy + self.connectionLimit = Host.normalizedPortForwardingMaxConnections(connectionLimit) self.makeReconnectController = makeReconnectController if let makeConnection { self.makeConnection = { _, connectionHandler, failureHandler in @@ -151,7 +147,7 @@ public final class PortForwardingController { } } else { self.makeConnection = { maximumConnections, connectionHandler, failureHandler in - if maximumConnections > 1 { + if poolPolicy == .hostConfigured { SSHForwardingConnectionPool( maximumConnections: maximumConnections, connectionHandler: connectionHandler, failureHandler: failureHandler, logger: logger) @@ -166,6 +162,30 @@ public final class PortForwardingController { public func state(for id: PortForwardingRule.ID) -> PortForwardingRuntimeState { states[id] ?? .stopped } public func connection(for id: PortForwardingRule.ID) -> PortForwardingConnection? { presentations[id] } + private func maximumConnections(for host: Host) -> Int { + switch poolPolicy { + case .singleConnection: 1 + case .hostConfigured: + min(connectionLimit, Host.normalizedPortForwardingMaxConnections(host.portForwardingMaxConnections)) + } + } + + /// The app supplies final connection limits; the transport has no commerce dependency. + public func updateConnectionLimit(_ maximum: Int) { + connectionLimit = Host.normalizedPortForwardingMaxConnections(maximum) + for context in connections.values { + if let host = hosts.host(context.hostID) { context.maximumConnections = maximumConnections(for: host) } + } + let previous = limitUpdateTask + limitUpdateTask = Task { [weak self] in + await previous?.value + guard let self else { return } + for context in Array(connections.values) { + await context.forwardingConnection?.updateMaximumConnections(context.maximumConnections) + } + } + } + public func start(_ id: PortForwardingRule.ID) { guard !state(for: id).isActive, let rule = hosts.portForwarding(id), rule.isValid, let host = hosts.host(rule.hostID), host.sshEnabled @@ -183,7 +203,7 @@ public final class PortForwardingController { } else { context = PortForwardingConnection( host: host, route: hosts.route(for: host), - maximumConnections: poolPolicy.maximumConnections(for: host), + maximumConnections: maximumConnections(for: host), reconnect: makeReconnectController()) connections[host.id] = context } @@ -396,6 +416,7 @@ public final class PortForwardingController { } let attemptSubmissions = silently ? [:] : context.submissions.takeAuthenticationSecrets() let attemptProxySubmissions = silently ? [:] : context.proxySubmissions.takeProxyAuthenticationSecrets() + context.maximumConnections = maximumConnections(for: host) let forwardingConnection = makeConnection( context.maximumConnections, { [weak self, weak context] id, generation, count in @@ -476,6 +497,7 @@ public final class PortForwardingController { context.additionalSnapshot = hosts.additionalConnectionSnapshot(from: configuration) } if let snapshot = context.additionalSnapshot { + await forwardingConnection.updateMaximumConnections(context.maximumConnections) await forwardingConnection.configureAdditionalConnections { [weak self] in guard let self else { throw CancellationError() } return try self.hosts.additionalConnectionRequest(from: snapshot) diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnection.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnection.swift index 498bd10f..6df18e43 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnection.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnection.swift @@ -29,6 +29,7 @@ public nonisolated struct SSHForwardingPoolStatus: Hashable, Sendable { public protocol SSHForwardingConnectionProtocol: Actor { func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws func configureAdditionalConnections(_ resolver: @escaping SSHAdditionalConnectionRequestResolver) async + func updateMaximumConnections(_ maximum: Int) async func connectionPoolStatus() async -> SSHForwardingPoolStatus func start(_ rule: PortForwardingRule, generation: UUID) async throws func stop(_ id: PortForwardingRule.ID) async @@ -37,6 +38,7 @@ public protocol SSHForwardingConnectionProtocol: Actor { extension SSHForwardingConnectionProtocol { public func configureAdditionalConnections(_ resolver: @escaping SSHAdditionalConnectionRequestResolver) async {} + public func updateMaximumConnections(_ maximum: Int) async {} public func connectionPoolStatus() async -> SSHForwardingPoolStatus { .single } } @@ -77,6 +79,9 @@ public actor SSHForwardingConnection: SSHForwardingConnectionProtocol { await worker.acceptShared(descriptor, ruleID: ruleID, generation: generation) } + /// Queried only after the pool has stopped assigning flows to this connection. + func hasActiveFlows() async -> Bool { await worker.hasActiveFlows() } + public func stop(_ id: PortForwardingRule.ID) async { await worker.stop(id) } @@ -333,6 +338,12 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { } } + func hasActiveFlows() async -> Bool { + await withCheckedContinuation { continuation in + queue.async { [self] in continuation.resume(returning: !flows.isEmpty) } + } + } + func stop(_ id: PortForwardingRule.ID) async { await withCheckedContinuation { continuation in queue.async { [self] in diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnectionPool.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnectionPool.swift index 13c199c2..32a5454d 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnectionPool.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnectionPool.swift @@ -22,7 +22,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { let generation: UUID } - private let maximumConnections: Int + private var maximumConnections: Int private let connectionHandler: SSHForwardingConnection.ConnectionHandler private let failureHandler: SSHForwardingConnection.FailureHandler private let logger: Logger @@ -34,6 +34,11 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { private var additionalRequest: SSHAdditionalConnectionRequestResolver? private var opening: [UUID: Slot] = [:] private var installing: [UUID: Slot] = [:] + private var openingTasks: [UUID: Task] = [:] + private var draining: Set = [] + private var assignments: [UUID: Int] = [:] + private var checkingIdle: Set = [] + private var recheckIdle: Set = [] private var retryTask: Task? private var stabilityTasks: [UUID: Task] = [:] private var retryAttempt = 0 @@ -46,7 +51,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { failureHandler: @escaping SSHForwardingConnection.FailureHandler, logger: Logger = ApplicationLogger.disabled("termind.port-forwarding-pool") ) { - self.maximumConnections = min(max(maximumConnections, 2), Host.portForwardingMaxConnectionsRange.upperBound) + self.maximumConnections = Host.normalizedPortForwardingMaxConnections(maximumConnections) self.connectionHandler = connectionHandler self.failureHandler = failureHandler self.logger = logger @@ -87,6 +92,28 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { SSHForwardingPoolStatus(connected: slots.count, target: desiredConnections) } + func updateMaximumConnections(_ maximum: Int) async { + maximumConnections = Host.normalizedPortForwardingMaxConnections(maximum) + rebalanceSlots() + // Pending auxiliaries have no accepted flows and can be cancelled immediately. + let pending: [Slot] + if slots.count + opening.count + installing.count > desiredConnections { + pending = Array(opening.values) + Array(installing.values) + opening.removeAll() + installing.removeAll() + for task in openingTasks.values { task.cancel() } + openingTasks.removeAll() + } else { + pending = [] + } + retryTask?.cancel() + retryTask = nil + publishPoolCounts() + for slot in pending { await slot.connection.disconnect() } + for id in Array(draining) { await retireIfIdle(id) } + fillPool() + } + func flowDistribution(for ruleID: PortForwardingRule.ID) async -> [Int] { slots.keys.sorted { $0.uuidString < $1.uuidString }.map { counts[$0]?[ruleID] ?? 0 } } @@ -99,6 +126,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { try await primary.connection.start(rule, generation: generation) primary.rules.insert(rule.id) } else { + // Draining slots may be retained again if the limit increases before their flows finish. for slot in slots.values { try await slot.connection.registerShared(rule, generation: generation) slot.rules.insert(rule.id) @@ -133,6 +161,12 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { retryTask = nil for task in stabilityTasks.values { task.cancel() } stabilityTasks.removeAll() + for task in openingTasks.values { task.cancel() } + openingTasks.removeAll() + draining.removeAll() + assignments.removeAll() + checkingIdle.removeAll() + recheckIdle.removeAll() let pending = Array(opening.values) + Array(installing.values) opening.removeAll() installing.removeAll() @@ -152,6 +186,40 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { private var needsPool: Bool { registrations.values.contains { $0.rule.kind != .remote } } private var desiredConnections: Int { needsPool ? maximumConnections : 1 } + private func rebalanceSlots() { + let auxiliaries = slots.keys.filter { $0 != primaryID }.sorted { $0.uuidString < $1.uuidString } + let retained = Set(auxiliaries.prefix(max(0, desiredConnections - 1))) + draining = Set(auxiliaries).subtracting(retained) + } + + private func retireIfIdle(_ id: UUID) async { + guard draining.contains(id), id != primaryID, assignments[id, default: 0] == 0, + let slot = slots[id] + else { return } + guard checkingIdle.insert(id).inserted else { + recheckIdle.insert(id) + return + } + let busy = await slot.connection.hasActiveFlows() + checkingIdle.remove(id) + let needsRecheck = recheckIdle.remove(id) != nil + if busy { + if needsRecheck { await retireIfIdle(id) } + return + } + // The worker query is authoritative: count callbacks may still be in flight. + guard draining.contains(id), id != primaryID, slots[id] === slot, + assignments[id, default: 0] == 0 + else { return } + slots[id] = nil + counts[id] = nil + draining.remove(id) + stabilityTasks.removeValue(forKey: id)?.cancel() + publishPoolCounts() + await slot.connection.disconnect() + fillPool() + } + private func makeSlot(id: UUID) -> Slot { let connection = SSHForwardingConnection( connectionHandler: { [weak self] ruleID, generation, count in @@ -173,10 +241,12 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { let slot = makeSlot(id: id) opening[id] = slot let resolver = additionalRequest! - Task { [weak self] in + openingTasks[id] = Task { [weak self] in do { let request = try await resolver() + try Task.checkCancellation() try await slot.connection.connect(request, progress: nil) + try Task.checkCancellation() await self?.opened(slot) } catch { await self?.openFailed(id: id, error: error) @@ -187,6 +257,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { } private func opened(_ slot: Slot) async { + defer { openingTasks[slot.id] = nil } guard opening.removeValue(forKey: slot.id) != nil, !disconnecting, slots.count + installing.count < desiredConnections else { @@ -221,12 +292,15 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { } private func openFailed(id: UUID, error: any Error) { + openingTasks[id] = nil guard opening.removeValue(forKey: id) != nil, !disconnecting else { return } scheduleRetry(for: error) } private func scheduleRetry(for error: any Error) { - guard !Self.requiresManualRetry(error), retryTask == nil, needsPool else { return } + guard !Self.requiresManualRetry(error), retryTask == nil, needsPool, + !disconnecting, slots.count + opening.count + installing.count < desiredConnections + else { return } let delays = [1, 2, 5, 10, 30] let delay = delays[min(retryAttempt, delays.count - 1)] retryAttempt += 1 @@ -258,15 +332,18 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { stabilityTasks.removeValue(forKey: slotID)?.cancel() counts[slotID] = nil slots[slotID] = nil + draining.remove(slotID) publishPoolCounts() if wasPrimary, let promoted = slots.values.first { primaryID = promoted.id + rebalanceSlots() do { for registration in registrations.values where registration.rule.kind == .remote { try await promoted.connection.start(registration.rule, generation: registration.generation) promoted.rules.insert(registration.rule.id) } fillPool() + for id in Array(draining) { await retireIfIdle(id) } } catch { failureHandler(error.localizedDescription) } @@ -281,6 +358,8 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { private func retireAuxiliaries() async { retryTask?.cancel() retryTask = nil + for task in openingTasks.values { task.cancel() } + openingTasks.removeAll() let pending = Array(opening.values) + Array(installing.values) opening.removeAll() installing.removeAll() @@ -289,6 +368,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { for slot in auxiliaries { slots[slot.id] = nil counts[slot.id] = nil + draining.remove(slot.id) stabilityTasks.removeValue(forKey: slot.id)?.cancel() } await withTaskGroup(of: Void.self) { group in @@ -302,7 +382,7 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { Darwin.close(descriptor) return } - let eligible = slots.values.filter { $0.rules.contains(ruleID) } + let eligible = slots.values.filter { $0.rules.contains(ruleID) && !draining.contains($0.id) } guard !eligible.isEmpty else { Darwin.close(descriptor) return @@ -313,16 +393,21 @@ actor SSHForwardingConnectionPool: SSHForwardingConnectionProtocol { } let slot = candidates[roundRobin % candidates.count] roundRobin &+= 1 - guard await slot.connection.acceptShared(descriptor, ruleID: ruleID, generation: generation) else { + assignments[slot.id, default: 0] += 1 + let accepted = await slot.connection.acceptShared(descriptor, ruleID: ruleID, generation: generation) + assignments[slot.id, default: 1] -= 1 + if assignments[slot.id] == 0 { assignments[slot.id] = nil } + if !accepted { Darwin.close(descriptor) - return } + await retireIfIdle(slot.id) } - private func receivedCount(_ count: Int, ruleID: UUID, generation: UUID, slotID: UUID) { + private func receivedCount(_ count: Int, ruleID: UUID, generation: UUID, slotID: UUID) async { guard registrations[ruleID]?.generation == generation, slots[slotID] != nil else { return } counts[slotID, default: [:]][ruleID] = count report(ruleID, generation: generation) + if count == 0 { await retireIfIdle(slotID) } } private func report(_ ruleID: UUID, generation: UUID?) { diff --git a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferCoordinator.swift b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferCoordinator.swift index 198e45f9..c588819e 100644 --- a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferCoordinator.swift +++ b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferCoordinator.swift @@ -159,8 +159,11 @@ public final class SFTPTransferCoordinator { globalLimit = limits.globalLimit perHostLimit = limits.perHostLimit strategy = limits.strategy - preferencesUpdateTask?.cancel() - preferencesUpdateTask = Task { await concurrencyScheduler.updatePreferences(limits) } + let previous = preferencesUpdateTask + preferencesUpdateTask = Task { + await previous?.value + await concurrencyScheduler.updatePreferences(limits) + } schedule() } diff --git a/TFI/App/MobileAppDependencies.swift b/TFI/App/MobileAppDependencies.swift index 6cf342ff..8858725b 100644 --- a/TFI/App/MobileAppDependencies.swift +++ b/TFI/App/MobileAppDependencies.swift @@ -13,6 +13,7 @@ struct MobileAppDependencies { var workspaceConnections: MobileWorkspaceConnectionServices var metrics: MobileMetricsServices var appReview: AppReviewController + var proAccess: ProAccessController? var diagnosis = DiagnosisController() var logging: any ApplicationLogManaging = DisabledApplicationLogManager.shared var connectionCoordinator: RemoteConnectionCoordinator = RemoteConnectionCoordinator() diff --git a/TFI/App/MobileAppModel.swift b/TFI/App/MobileAppModel.swift index bf59b108..1807f960 100644 --- a/TFI/App/MobileAppModel.swift +++ b/TFI/App/MobileAppModel.swift @@ -217,7 +217,7 @@ final class MobileAppModel { self.defaults = inMemory || hosts != nil ? nil : defaults snippetTimestampFallback = MobileSnippetTimestamp.fallback(in: self.defaults) liveActivities = dependencies.liveActivities ?? MobileLiveActivityController(enabled: !inMemory && hosts == nil) - pro = inMemory || hosts != nil ? .unrestrictedForTesting() : .live() + pro = dependencies.proAccess ?? (inMemory || hosts != nil ? .unrestrictedForTesting() : .live()) let persistence = HostPersistence(context: vaultContext) _ = try persistence.load() let secrets: any SecretStoring = @@ -316,7 +316,8 @@ final class MobileAppModel { service: AppleSpeechTranscriptionService(audioInput: speechAudio) ) speechAudio.onInterruption = { [weak speech] in speech?.cancelAll() } - let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences) + let sftpPreferences = pro.effectiveSFTPPreferences(sftpSettings.preferences) + let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpPreferences) assistant = MobileAssistantModel( configuration: ai, repository: MemoryIndexedAgentRepository( @@ -342,9 +343,9 @@ final class MobileAppModel { forwardings = dependencies.makeForwardings(library) transfers = SFTPTransferCoordinator( store: library, clientFactory: { connectionCoordinator.makeSFTPClient() }, - globalLimit: sftpSettings.preferences.globalLimit, - perHostLimit: sftpSettings.preferences.perHostLimit, - strategy: sftpSettings.preferences.strategy, + globalLimit: sftpPreferences.globalLimit, + perHostLimit: sftpPreferences.perHostLimit, + strategy: sftpPreferences.strategy, concurrencyScheduler: sftpConcurrencyScheduler) sftpEdits = SFTPEditService(transfers: transfers, host: { [weak library] in library?.host($0) }) library.willDeleteHost = { [weak sftpEdits, weak forwardings] hostID, _ in @@ -391,8 +392,13 @@ final class MobileAppModel { connectionHistorySync?.didApplyRemote = { [weak self] _ in self?.connectionHistory.reload() } deviceSync?.didApplyRemote = { [weak self] _ in self?.devices.reload() } vaultSync?.didApplyRemote = { [weak self] in self?.applyRemoteVault($0, keys: $1) } - sftpSettings.didChange = { [weak transfers] preferences in - transfers?.updatePreferences(preferences) + sftpSettings.didChange = { [weak transfers, weak pro] preferences in + guard let pro else { return } + transfers?.updatePreferences(pro.effectiveSFTPPreferences(preferences)) + } + pro.accessDidChange = { [weak self] in + guard let self else { return } + transfers.updatePreferences(pro.effectiveSFTPPreferences(sftpSettings.preferences)) } sshSettings.didChange = { preferences in connectionCoordinator.updateSSHPreferences(preferences) diff --git a/TFI/Features/Settings/MobileSFTPSettingsView.swift b/TFI/Features/Settings/MobileSFTPSettingsView.swift index 97e18d8e..a512ca8a 100644 --- a/TFI/Features/Settings/MobileSFTPSettingsView.swift +++ b/TFI/Features/Settings/MobileSFTPSettingsView.swift @@ -5,11 +5,12 @@ import TermindPresentation struct MobileConnectionSettingsView: View { let ssh: SSHSettingsModel let sftp: SFTPSettingsModel + var sftpAccess = MultiConnectionSettingsAccess(isAllowed: true, requestAccess: {}) var body: some View { Form { Section(AppStrings.Vault.ssh.swiftUIKey) { SSHSettingsFields(model: ssh) } - Section(AppStrings.SFTP.sftp.swiftUIKey) { SFTPSettingsFields(model: sftp) } + Section(AppStrings.SFTP.sftp.swiftUIKey) { SFTPSettingsFields(model: sftp, access: sftpAccess) } } .navigationTitle(AppStrings.Settings.connections.swiftUIKey) } diff --git a/TFI/Features/Settings/MobileSettingsView.swift b/TFI/Features/Settings/MobileSettingsView.swift index 5e34b103..09047ba2 100644 --- a/TFI/Features/Settings/MobileSettingsView.swift +++ b/TFI/Features/Settings/MobileSettingsView.swift @@ -41,7 +41,11 @@ struct MobileSettingsView: View { Label(AppStrings.Settings.terminal.swiftUIKey, systemImage: "terminal") } NavigationLink { - MobileConnectionSettingsView(ssh: app.sshSettings, sftp: app.sftpSettings) + MobileConnectionSettingsView( + ssh: app.sshSettings, sftp: app.sftpSettings, + sftpAccess: .init(isAllowed: app.pro.canUse(.sftpMultiConnection)) { + app.pro.require(.sftpMultiConnection) + }) } label: { Label(AppStrings.Settings.connections.swiftUIKey, systemImage: "network") } diff --git a/TFITests/MobileMultiConnectionAccessTests.swift b/TFITests/MobileMultiConnectionAccessTests.swift new file mode 100644 index 00000000..11c826c5 --- /dev/null +++ b/TFITests/MobileMultiConnectionAccessTests.swift @@ -0,0 +1,37 @@ +import TermindDomain +import Testing + +@testable import TermindIOS +@testable import TermindInfrastructure + +@Suite @MainActor +struct MobileMultiConnectionAccessTests { + @Test(arguments: SFTPConcurrencyStrategy.allCases) + func entitlementChangesPreserveSavedSettingsWithoutAnyScene(_ strategy: SFTPConcurrencyStrategy) async throws { + let service = MobileConnectionPurchaseService() + let pro = ProAccessController(purchaseService: service, observeUpdates: false) + var dependencies = MobileAppDependencies.testing() + dependencies.proAccess = pro + let app = try MobileAppModel(dependencies: dependencies) + app.sftpSettings.setStrategy(strategy) + #expect(app.transfers.strategy == .channelsOnly) + service.status = .purchased(.purchased) + await pro.refresh() + #expect(app.transfers.strategy == strategy) + service.status = .free(trialAvailable: false) + await pro.refresh() + #expect(app.transfers.strategy == .channelsOnly) + #expect(app.sftpSettings.preferences.strategy == strategy) + app.sftpSettings.set(60, for: \.globalLimit) + #expect(app.transfers.globalLimit == 60) + #expect(app.transfers.strategy == .channelsOnly) + #expect(!pro.isPaywallPresented) + } +} + +@MainActor +private final class MobileConnectionPurchaseService: ProPurchaseService { + let updates = AsyncStream { $0.finish() } + var status = EntitlementStatus.free(trialAvailable: false) + func currentStatus() async throws -> EntitlementStatus { status } +} diff --git a/TFM/App/AppModel.swift b/TFM/App/AppModel.swift index f1f1bd32..8e6f305e 100644 --- a/TFM/App/AppModel.swift +++ b/TFM/App/AppModel.swift @@ -244,7 +244,8 @@ final class AppModel { store: dependencies.assistantModelSelectionStore ?? InMemoryAssistantModelSelectionStore()) assistantChatCollaborationStore = dependencies.assistantChatCollaborationStore ?? InMemoryAssistantChatCollaborationStore() - let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences) + let sftpPreferences = proAccess.effectiveSFTPPreferences(sftpSettings.preferences) + let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpPreferences) let agentHostServices = AgentHostServices.live( store: hosts, coordinator: connectionCoordinator, @@ -261,13 +262,15 @@ final class AppModel { metricsSync = dependencies.metricsSync forwarding = PortForwardingController( hosts: hosts, poolPolicy: .hostConfigured, + connectionLimit: proAccess.effectiveForwardingConnections( + Host.portForwardingMaxConnectionsRange.upperBound), logger: dependencies.logging.makeLogger(label: "termind.port-forwarding")) sftpTransfers = SFTPTransferCoordinator( store: hosts, clientFactory: { connectionCoordinator.makeSFTPClient() }, - globalLimit: sftpSettings.preferences.globalLimit, - perHostLimit: sftpSettings.preferences.perHostLimit, - strategy: sftpSettings.preferences.strategy, + globalLimit: sftpPreferences.globalLimit, + perHostLimit: sftpPreferences.perHostLimit, + strategy: sftpPreferences.strategy, concurrencyScheduler: sftpConcurrencyScheduler ) sftpEdits = SFTPEditService(transfers: sftpTransfers, host: { [weak hosts] in hosts?.host($0) }) @@ -374,8 +377,15 @@ final class AppModel { vaultSync?.didApplyRemote = { [weak self] snapshot, keys in self?.applyRemoteVaultSnapshot(snapshot, changedKeys: keys) } - sftpSettings.didChange = { [weak sftpTransfers] preferences in - sftpTransfers?.updatePreferences(preferences) + sftpSettings.didChange = { [weak sftpTransfers, weak proAccess] preferences in + guard let proAccess else { return } + sftpTransfers?.updatePreferences(proAccess.effectiveSFTPPreferences(preferences)) + } + proAccess.accessDidChange = { [weak self] in + guard let self else { return } + self.sftpTransfers.updatePreferences(self.proAccess.effectiveSFTPPreferences(self.sftpSettings.preferences)) + self.forwarding.updateConnectionLimit( + self.proAccess.effectiveForwardingConnections(Host.portForwardingMaxConnectionsRange.upperBound)) } sshSettings.didChange = { preferences in connectionCoordinator.updateSSHPreferences(preferences) diff --git a/TFM/Features/HostEditor/HostEditorView.swift b/TFM/Features/HostEditor/HostEditorView.swift index 8e7274a1..b6662cd8 100644 --- a/TFM/Features/HostEditor/HostEditorView.swift +++ b/TFM/Features/HostEditor/HostEditorView.swift @@ -295,13 +295,15 @@ struct HostEditorView: View { Section(AppStrings.Navigation.portForwarding.swiftUIKey) { Stepper( - value: $draft.portForwardingMaxConnections, + value: forwardingConnections, in: Host.portForwardingMaxConnectionsRange ) { HStack { - Text(AppStrings.Vault.maxConnections) + Text( + app.proAccess.canUse(.portForwardingMultiConnection) + ? AppStrings.Vault.maxConnections : AppStrings.Pro.maxConnections) Spacer() - Text(draft.portForwardingMaxConnections, format: .number).monospacedDigit() + Text(forwardingConnections.wrappedValue, format: .number).monospacedDigit() } } Text(AppStrings.Vault.forwardingPoolExplanation) @@ -515,6 +517,12 @@ struct HostEditorView: View { // MARK: - Picking, and making + private var forwardingConnections: Binding { + MultiConnectionSettingsAccess(isAllowed: app.proAccess.canUse(.portForwardingMultiConnection)) { + app.proAccess.require(.portForwardingMultiConnection) + }.forwardingConnections($draft.portForwardingMaxConnections) + } + /// What a picker row means: nothing, one of the saved records, or the row /// that makes a new one. /// diff --git a/TFM/Features/Settings/SettingsPageView.swift b/TFM/Features/Settings/SettingsPageView.swift index 3ba1ff77..f1feb9b2 100644 --- a/TFM/Features/Settings/SettingsPageView.swift +++ b/TFM/Features/Settings/SettingsPageView.swift @@ -583,7 +583,13 @@ private struct ConnectionSettings: View { var body: some View { Section(AppStrings.Vault.ssh.swiftUIKey) { SSHSettingsFields(model: app.sshSettings) } - Section(AppStrings.SFTP.sftp.swiftUIKey) { SFTPSettingsFields(model: app.sftpSettings) } + Section(AppStrings.SFTP.sftp.swiftUIKey) { + SFTPSettingsFields( + model: app.sftpSettings, + access: .init(isAllowed: app.proAccess.canUse(.sftpMultiConnection)) { + app.proAccess.require(.sftpMultiConnection) + }) + } } } diff --git a/TFMTests/PortForwardingReconnectTests.swift b/TFMTests/PortForwardingReconnectTests.swift index 38eed145..495003b7 100644 --- a/TFMTests/PortForwardingReconnectTests.swift +++ b/TFMTests/PortForwardingReconnectTests.swift @@ -8,6 +8,23 @@ import struct TermindDomain.Host @Suite @MainActor struct PortForwardingReconnectTests { + @Test func numericLimitsUpdateLiveConnectionsAndSurviveRecoveryWithoutChangingHosts() async throws { + let fixture = ReconnectFixture(poolPolicy: .hostConfigured, connectionLimit: 1, hostConnections: 8) + defer { fixture.stop() } + try await fixture.start([0, 1, 2]) + #expect(await fixture.sessions[0].maximumConnections == 1) + fixture.controller.updateConnectionLimit(64) + try await eventually { await fixture.sessions[0].maximumConnections == 8 } + fixture.controller.updateConnectionLimit(1) + try await eventually { await fixture.sessions[0].maximumConnections == 1 } + #expect(fixture.library.host(fixture.host.id)?.portForwardingMaxConnections == 8) + await fixture.sessions[0].fail() + try await fixture.retry() + try await fixture.waitForRunning([0, 1, 2]) + #expect(await fixture.sessions[1].maximumConnections == 1) + #expect(!fixture.context!.journey.isPresented) + } + @Test func compactLabelsHideConnectionCountsAndFailureDetails() { let running = PortForwardingRuntimeState.running( activeConnections: 37, sshConnections: 3, targetSSHConnections: 8) @@ -376,6 +393,8 @@ private final class ReconnectFixture { var sessions: [ReconnectSession] = [] var nextError: TermindSSHError? let modes: [ReconnectSession.Mode] + let poolPolicy: PortForwardingController.PoolPolicy + let connectionLimit: Int lazy var controller = makeController() private func makeController() -> PortForwardingController { @@ -387,17 +406,25 @@ private final class ReconnectFixture { return session } return PortForwardingController( - hosts: library, makeConnection: makeConnection, + hosts: library, poolPolicy: poolPolicy, connectionLimit: connectionLimit, makeConnection: makeConnection, makeReconnectController: { [gate] in SessionReconnectController(policy: .sessionDefault, sleep: { await gate.sleep($0) }, now: Date.init) }) } var context: PortForwardingConnection? { controller.connection(for: rules[0].id) } - init(modes: [ReconnectSession.Mode] = [], usesJump: Bool = false, trusted: Bool = true) { + init( + modes: [ReconnectSession.Mode] = [], usesJump: Bool = false, trusted: Bool = true, + poolPolicy: PortForwardingController.PoolPolicy = .singleConnection, connectionLimit: Int = 64, + hostConnections: Int = 1 + ) { self.modes = modes + self.poolPolicy = poolPolicy + self.connectionLimit = connectionLimit jump = Host(name: "Jump", host: "jump.invalid", identityID: identity.id) - host = Host(name: "Host", host: "host.invalid", identityID: identity.id, jumpHostID: usesJump ? jump.id : nil) + host = Host( + name: "Host", host: "host.invalid", identityID: identity.id, jumpHostID: usesJump ? jump.id : nil, + portForwardingMaxConnections: hostConnections) rules = [ PortForwardingRule(hostID: host.id, kind: .local, bindPort: 8000), PortForwardingRule(hostID: host.id, kind: .remote, bindPort: 9000), @@ -482,6 +509,8 @@ private actor ReconnectSession: SSHForwardingConnectionProtocol { var generations: [UUID: UUID] = [:] var hasProgress = false var disconnects = 0 + var maximumConnections = 1 + func updateMaximumConnections(_ maximum: Int) async { maximumConnections = maximum } private var connectGate: CheckedContinuation? private var disconnectGate: CheckedContinuation? private var pendingStarts: [UUID: CheckedContinuation] = [:] diff --git a/TFMTests/SFTPConcurrencySchedulerTests.swift b/TFMTests/SFTPConcurrencySchedulerTests.swift index 1420be2f..90bf79de 100644 --- a/TFMTests/SFTPConcurrencySchedulerTests.swift +++ b/TFMTests/SFTPConcurrencySchedulerTests.swift @@ -5,6 +5,30 @@ import Testing @Suite struct SFTPConcurrencySchedulerTests { + @Test func finalPreferencesAffectQueuedWorkWithoutChangingGrantedPermits() async throws { + let scheduler = SFTPConcurrencyScheduler( + preferences: .init(globalLimit: 1, perHostLimit: 1, strategy: .connectionsPreferred)) + let host = UUID() + let first = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: .workspace(UUID()), byteCount: 80_000_000) + let queued = Task { + try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: .agentRun(UUID()), byteCount: 80_000_000) + } + try await eventually { await scheduler.queuedCount == 1 } + await scheduler.updatePreferences(.init(globalLimit: 1, perHostLimit: 1, strategy: .channelsOnly)) + #expect(first.schedulingMode == .connectionsPreferred) + await first.release() + let next = try await queued.value + #expect(next.schedulingMode == .channelsOnly) + await next.release() + await scheduler.updatePreferences(.init(strategy: .smart)) + let upgraded = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: .workspace(UUID()), byteCount: 80_000_000) + #expect(upgraded.schedulingMode == .connectionsPreferred) + await upgraded.release() + } + @Test func smartResolvesTheSixtyFourMBBoundaryAndUnknownSizeToSessions() async throws { let scheduler = SFTPConcurrencyScheduler( preferences: SFTPPreferences(globalLimit: 3, perHostLimit: 3, strategy: .smart)) diff --git a/TFMTests/SFTPSettingsIntegrationTests.swift b/TFMTests/SFTPSettingsIntegrationTests.swift index d5568d5c..099fd6f3 100644 --- a/TFMTests/SFTPSettingsIntegrationTests.swift +++ b/TFMTests/SFTPSettingsIntegrationTests.swift @@ -2,12 +2,88 @@ import AppKit import SwiftUI import TermindApplication import TermindDomain +import TermindInfrastructure +import TermindPresentation import Testing @testable import Termind +@testable import TermindInfrastructure @Suite(.serialized) @MainActor struct SFTPSettingsIntegrationTests { + @Test(arguments: ["en", "zh-Hans", "zh-Hant"]) + func freeSettingsRenderOffscreenWithoutMutatingTheSavedChoice(language: String) async throws { + let settings = SFTPSettingsModel() + settings.setStrategy(.connectionsPreferred) + let view = Form { + SFTPSettingsFields(model: settings, access: .init(isAllowed: false, requestAccess: {})) + } + .formStyle(.grouped) + .environment(\.locale, Locale(identifier: language)) + .frame(width: 620, height: 340) + let host = NSHostingView(rootView: view) + host.frame = NSRect(x: 0, y: 0, width: 620, height: 340) + host.appearance = NSAppearance(named: .aqua) + host.layoutSubtreeIfNeeded() + try await Task.sleep(for: .milliseconds(100)) + let bitmap = try #require(host.bitmapImageRepForCachingDisplay(in: host.bounds)) + host.cacheDisplay(in: host.bounds, to: bitmap) + let data = try #require(bitmap.representation(using: .png, properties: [:])) + Attachment.record(data, named: "free-sftp-settings-\(language).png") + #expect(data.count > 2_000) + #expect(settings.preferences.strategy == .connectionsPreferred) + } + + @Test(arguments: SFTPConcurrencyStrategy.allCases) + func purchaseRestoresSavedStrategyWithoutOpeningSettings(_ strategy: SFTPConcurrencyStrategy) async { + let service = ConnectionPurchaseService() + let pro = ProAccessController(purchaseService: service, observeUpdates: false) + let settings = SFTPSettingsModel() + settings.setStrategy(strategy) + let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings, proAccess: pro) + defer { app.shutdownForTermination() } + #expect(app.sftpTransfers.strategy == .channelsOnly) + settings.set(60, for: \.globalLimit) + service.status = .purchased(.purchased) + await pro.refresh() + #expect(app.sftpTransfers.strategy == strategy) + #expect(app.sftpTransfers.globalLimit == 60) + #expect(settings.preferences.strategy == strategy) + #expect(!pro.isPaywallPresented) + } + @Test(arguments: SFTPConcurrencyStrategy.allCases) + func losingAccessChangesRuntimeWithoutRewritingPreferencesOrOpeningPaywall(_ strategy: SFTPConcurrencyStrategy) + async + { + let settings = SFTPSettingsModel() + settings.setStrategy(strategy) + let pro = ProAccessController.testing(status: .purchased(.purchased)) + let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings, proAccess: pro) + defer { app.shutdownForTermination() } + #expect(app.sftpTransfers.strategy == strategy) + await pro.refresh() + #expect(app.sftpTransfers.strategy == .channelsOnly) + #expect(settings.preferences.strategy == strategy) + #expect(!pro.isPaywallPresented) + settings.set(50, for: \.globalLimit) + #expect(app.sftpTransfers.strategy == .channelsOnly) + #expect(app.sftpTransfers.globalLimit == 50) + #expect(settings.preferences.strategy == strategy) + } + + @Test func freeStartupAndSyncedPreferencesUseOnlyTheEffectiveStrategy() { + let settings = SFTPSettingsModel() + let pro = ProAccessController.testing(status: .free(trialAvailable: true)) + let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings, proAccess: pro) + defer { app.shutdownForTermination() } + #expect(settings.preferences.strategy == .smart) + #expect(app.sftpTransfers.strategy == .channelsOnly) + settings.applyRemoteSnapshot(.init(strategy: .connectionsPreferred)) + #expect(settings.preferences.strategy == .connectionsPreferred) + #expect(app.sftpTransfers.strategy == .channelsOnly) + #expect(!pro.isPaywallPresented) + } + @Test func localAndRemoteChangesReachTheQueueWithoutOpeningSettings() { let settings = SFTPSettingsModel() let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings) @@ -60,3 +136,10 @@ struct SFTPSettingsIntegrationTests { } } } + +@MainActor +private final class ConnectionPurchaseService: ProPurchaseService { + let updates = AsyncStream { $0.finish() } + var status = EntitlementStatus.free(trialAvailable: false) + func currentStatus() async throws -> EntitlementStatus { status } +} diff --git a/TFMTests/SSHForwardingWireTests.swift b/TFMTests/SSHForwardingWireTests.swift index c7c0b613..216fd63d 100644 --- a/TFMTests/SSHForwardingWireTests.swift +++ b/TFMTests/SSHForwardingWireTests.swift @@ -9,6 +9,76 @@ import Testing /// Uses the loopback Paramiko or forwarding-only OpenSSH fixture in Scripts/. @Suite(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_SSH_FIXTURE"] != nil), .serialized) nonisolated struct SSHForwardingWireTests { + @Test func poolShrinksWithoutClosingExistingFlowsAndCanExpandAgain() async throws { + let fixture = try fixture() + let pool = SSHForwardingConnectionPool( + maximumConnections: 1, connectionHandler: { _, _, _ in }, + failureHandler: { Issue.record(Comment(rawValue: $0)) }) + try await pool.connect(request(fixture, jumps: 0), progress: nil) + await pool.configureAdditionalConnections { request(fixture, jumps: 0).enforcingInteractionPolicy() } + let local = try rule(.local, fixture: fixture) + try await pool.start(local, generation: UUID()) + defer { Task { await pool.disconnect() } } + #expect(await pool.connectionPoolStatus() == .single) + await pool.updateMaximumConnections(3) + try await waitForPool(pool, connected: 3, target: 3) + var clients: [Int32] = [] + defer { for client in clients { Darwin.close(client) } } + for _ in 0..<12 { clients.append(try SSHSocketConnector.open(host: "127.0.0.1", port: local.bindPort)) } + for _ in 0..<1_000 { + if await pool.flowDistribution(for: local.id).reduce(0, +) == clients.count { break } + try await Task.sleep(for: .milliseconds(5)) + } + #expect(await pool.flowDistribution(for: local.id).filter { $0 > 0 }.count > 1) + await pool.updateMaximumConnections(1) + #expect(await pool.connectionPoolStatus().target == 1) + // Every established TCP stream must still pass data after the downgrade. + for client in clients { + var sent: UInt8 = 42 + #expect(Darwin.write(client, &sent, 1) == 1) + var timeout = timeval(tv_sec: 5, tv_usec: 0) + setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, &timeout, socklen_t(MemoryLayout.size)) + var received: UInt8 = 0 + #expect(Darwin.read(client, &received, 1) == 1) + #expect(received == sent) + } + // A new stream still works on the retained connection while auxiliaries drain. + try await exchange(port: local.bindPort, target: nil, seed: 72, rounds: 2) + let addedWhileDraining = try rule(.local, fixture: fixture) + try await pool.start(addedWhileDraining, generation: UUID()) + await pool.updateMaximumConnections(3) + try await waitForPool(pool, connected: 3, target: 3) + var addedClients: [Int32] = [] + defer { for client in addedClients { Darwin.close(client) } } + for _ in 0..<12 { + addedClients.append(try SSHSocketConnector.open(host: "127.0.0.1", port: addedWhileDraining.bindPort)) + } + for _ in 0..<1_000 { + if await pool.flowDistribution(for: addedWhileDraining.id).filter({ $0 > 0 }).count > 1 { break } + try await Task.sleep(for: .milliseconds(5)) + } + #expect(await pool.flowDistribution(for: addedWhileDraining.id).filter { $0 > 0 }.count > 1) + for client in addedClients { Darwin.close(client) } + addedClients.removeAll() + await pool.updateMaximumConnections(1) + for client in clients { Darwin.close(client) } + clients.removeAll() + try await waitForPool(pool, connected: 1, target: 1) + await pool.updateMaximumConnections(3) + try await waitForPool(pool, connected: 3, target: 3) + await pool.updateMaximumConnections(1) + try await waitForPool(pool, connected: 1, target: 1) + } + + private func waitForPool(_ pool: SSHForwardingConnectionPool, connected: Int, target: Int) async throws { + let expected = SSHForwardingPoolStatus(connected: connected, target: target) + for _ in 0..<1_000 { + if await pool.connectionPoolStatus() == expected { return } + try await Task.sleep(for: .milliseconds(5)) + } + #expect(await pool.connectionPoolStatus() == expected) + } + @Test(arguments: [PortForwardingRule.Kind.local, .dynamic]) func disconnectReleasesThePortBeforeReturningEvenWithEstablishedStreams(_ kind: PortForwardingRule.Kind) async throws