Unified SFTP scheduler

This commit is contained in:
Kaede Akino committed 2026-09-19 22:42:15 +08:00
1 parent d9616b7b77
commit 145cfd535a
29 files changed
+1089 -178

No files matched your search

+7 -4
View File
@@ -15,7 +15,7 @@ struct SFTPSettingsSyncTests {
func importsExistingValuesOnlyOnce(global: Int?) throws {
let fixture = try Fixture(global: global, perHost: global == nil ? nil : 2)
defer { fixture.cleanUp() }
let expected = SFTPPreferences(globalLimit: global ?? 15, perHostLimit: global == nil ? 5 : 2)
let expected = SFTPPreferences(globalLimit: global ?? 30, perHostLimit: global == nil ? 10 : 2)
#expect(try fixture.store.loadSFTP() == expected)
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == expected.globalLimit)
let original = try #require(try fixture.store.journal.envelope(for: key))
@@ -52,7 +52,7 @@ struct SFTPSettingsSyncTests {
}
var localChanges: [SettingsMutationSet] = []
source.store.didRecordChanges = { localChanges.append($0) }
let value = SFTPPreferences(globalLimit: 100, perHostLimit: 50)
let value = SFTPPreferences(globalLimit: 100, perHostLimit: 50, strategy: .connectionsPreferred)
try source.store.saveSFTP(value)
#expect(localChanges.last?.keys == [key])
let outgoing = try #require(try source.store.journal.envelope(for: key))
@@ -71,6 +71,9 @@ struct SFTPSettingsSyncTests {
#expect(try destination.store.loadSFTP() == value)
#expect(destination.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 100)
#expect(destination.defaults.integer(forKey: SettingsKey.sftpPerHostTransferLimit) == 50)
#expect(
destination.defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy)
== SFTPConcurrencyStrategy.connectionsPreferred.rawValue)
#expect(callbacks == 0)
#expect(result.acceptedKeys == [key])
#expect(try !destination.store.journal.pendingEnvelopes().contains { $0.key == key })
@@ -122,7 +125,7 @@ struct SFTPSettingsSyncTests {
#expect(try SettingsSyncCoordinator.payloadData(from: record) == bytes)
#expect(throws: SettingsSyncPersistenceError.self) { try fixture.store.applyRemote([invalid]) }
#expect(try fixture.store.loadSFTP() == SFTPPreferences())
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 15)
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 30)
}
@Test func cloudRecordTypeMustMatchItsPayloadAndRestoredSystemFields() throws {
@@ -154,7 +157,7 @@ struct SFTPSettingsSyncTests {
#expect(throws: (any Error).self) { try fixture.store.saveSFTP(.init(globalLimit: 100, perHostLimit: 50)) }
#expect(try fixture.store.loadSFTP() == SFTPPreferences())
#expect(try fixture.store.journal.envelope(for: key) == before)
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 15)
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 30)
#expect(callbacks == 0)
}
+23 -4
View File
@@ -6,9 +6,9 @@ import Testing
@Suite @MainActor
struct SFTPSettingsTests {
@Test func defaultsAndRangesHaveOneSharedDefinition() {
#expect(SFTPPreferences() == SFTPPreferences(globalLimit: 15, perHostLimit: 5))
#expect(SettingsKey.defaultSFTPGlobalTransferLimit == 15)
#expect(SettingsKey.defaultSFTPPerHostTransferLimit == 5)
#expect(SFTPPreferences() == SFTPPreferences(globalLimit: 30, perHostLimit: 10, strategy: .smart))
#expect(SettingsKey.defaultSFTPGlobalTransferLimit == 30)
#expect(SettingsKey.defaultSFTPPerHostTransferLimit == 10)
#expect(SettingsKey.sftpGlobalTransferLimitRange == 1...100)
#expect(SettingsKey.sftpPerHostTransferLimitRange == 1...50)
}
@@ -37,7 +37,7 @@ struct SFTPSettingsTests {
persistence.fail = false
#expect(model.set(100, for: \.globalLimit))
#expect(model.storageError == nil)
#expect(received == [SFTPPreferences(globalLimit: 100, perHostLimit: 5)])
#expect(received == [SFTPPreferences(globalLimit: 100, perHostLimit: 10)])
}
@Test func remoteChangesNotifyOnceWithoutSavingAndLocalChangesCommitThePair() {
@@ -55,6 +55,25 @@ struct SFTPSettingsTests {
#expect(persistence.saves == 1)
#expect(received.last == persistence.value)
}
@Test func strategyPersistsAndOlderJSONDefaultsToSmart() throws {
let legacy = try JSONDecoder().decode(
SFTPPreferences.self,
from: Data(#"{"globalLimit":15,"perHostLimit":5}"#.utf8))
#expect(legacy == SFTPPreferences(globalLimit: 15, perHostLimit: 5, strategy: .smart))
let persistence = TestSFTPPreferencesPersistence()
let model = SFTPSettingsModel(persistence: persistence)
#expect(model.setStrategy(.channelsOnly))
#expect(persistence.value.strategy == .channelsOnly)
#expect(persistence.saves == 1)
#expect(throws: DecodingError.self) {
_ = try JSONDecoder().decode(
SFTPPreferences.self,
from: Data(#"{"globalLimit":30,"perHostLimit":10,"strategy":"future"}"#.utf8))
}
}
}
@MainActor
+88 -3
View File
@@ -4486,19 +4486,19 @@
}
}
},
"Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, the session transfers files one at a time." : {
"Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, transfers use parallel channels on the existing connection." : {
"extractionState" : "manual",
"localizations" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "更改会应用到排队中的传输。正在进行的传输不受影响。如果额外连接需要身份验证提示,会话将逐个传输文件。"
"value" : "更改会应用到排队中的传输。正在进行的传输不受影响。如果额外连接无法在不显示身份验证提示的情况下打开,传输将改用现有连接上的并行通道。"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "更改會套用至佇列中的傳輸。正在進行的傳輸不受影響。如果額外連線需要驗證提示,工作階段將逐一傳輸檔案。"
"value" : "更改會套用至佇列中的傳輸。正在進行的傳輸不受影響。如果額外連線無法在不顯示驗證提示的情況下開啟,傳輸將改用現有連線上的平行通道。"
}
}
}
@@ -5667,6 +5667,57 @@
}
}
},
"Channels Only" : {
"extractionState" : "manual",
"localizations" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "仅通道"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "僅通道"
}
}
}
},
"Concurrent Strategy" : {
"extractionState" : "manual",
"localizations" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "并发策略"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "並行策略"
}
}
}
},
"Connections Preferred" : {
"extractionState" : "manual",
"localizations" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "连接优先"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "連線優先"
}
}
}
},
"Concurrent Transfers" : {
"extractionState" : "manual",
"localizations" : {
@@ -5684,6 +5735,40 @@
}
}
},
"Smart" : {
"extractionState" : "manual",
"localizations" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "智能"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "智慧"
}
}
}
},
"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" : {
"zh-Hans" : {
"stringUnit" : {
"state" : "translated",
"value" : "智能模式对不超过64 MB的文件使用通道,对更大的文件优先使用额外连接。大小未知的传输使用通道。"
}
},
"zh-Hant" : {
"stringUnit" : {
"state" : "translated",
"value" : "智慧模式對不超過64 MB的檔案使用通道,對更大的檔案優先使用額外連線。大小未知的傳輸使用通道。"
}
}
}
},
"Configure Model Providers" : {
"localizations" : {
"zh-Hans" : {
@@ -32,6 +32,21 @@ public final class SFTPSettingsModel {
}
}
@discardableResult
public func setStrategy(_ value: SFTPConcurrencyStrategy) -> Bool {
var next = preferences
next.strategy = value
guard next != preferences else { return true }
do {
try persistence?.saveSFTP(next)
applyRemoteSnapshot(next)
return true
} catch {
storageError = error.localizedDescription
return false
}
}
public func applyRemoteSnapshot(_ value: SFTPPreferences) {
let next = value.normalized
let changed = preferences != next
@@ -29,6 +29,7 @@ public nonisolated enum SettingsKey {
public static let settingsSyncEnabled = "settings.sync.enabled"
public static let settingsSyncEnvironment = "settings.sync.environment"
public static let shellIntegration = "terminal.shellIntegration"
public static let sftpConcurrencyStrategy = "sftp.transfers.strategy"
public static let sftpGlobalTransferLimit = "sftp.transfers.globalLimit"
public static let sftpPerHostTransferLimit = "sftp.transfers.perHostLimit"
public static let defaultSFTPGlobalTransferLimit = SFTPPreferences.defaultGlobalLimit
@@ -1,23 +1,39 @@
import Foundation
public nonisolated enum SFTPConcurrencyStrategy: String, Codable, CaseIterable, Hashable, Sendable {
case smart
case channelsOnly = "channels_only"
case connectionsPreferred = "connections_preferred"
public static let smartConnectionThreshold: Int64 = 64_000_000
}
/// File-transfer concurrency shared by terminal uploads and SFTP browsers.
public nonisolated struct SFTPPreferences: Codable, Hashable, Sendable {
public static let defaultGlobalLimit = 15
public static let defaultPerHostLimit = 5
public static let defaultGlobalLimit = 30
public static let defaultPerHostLimit = 10
public static let globalLimitRange = 1...100
public static let perHostLimitRange = 1...50
public var globalLimit: Int
public var perHostLimit: Int
public var strategy: SFTPConcurrencyStrategy
public init(globalLimit: Int = defaultGlobalLimit, perHostLimit: Int = defaultPerHostLimit) {
public init(
globalLimit: Int = defaultGlobalLimit,
perHostLimit: Int = defaultPerHostLimit,
strategy: SFTPConcurrencyStrategy = .smart
) {
self.globalLimit = min(max(globalLimit, Self.globalLimitRange.lowerBound), Self.globalLimitRange.upperBound)
self.perHostLimit = min(
max(perHostLimit, Self.perHostLimitRange.lowerBound),
Self.perHostLimitRange.upperBound, self.globalLimit)
self.strategy = strategy
}
public var normalized: Self { Self(globalLimit: globalLimit, perHostLimit: perHostLimit) }
public var normalized: Self {
Self(globalLimit: globalLimit, perHostLimit: perHostLimit, strategy: strategy)
}
/// Decoded remote values must be validated before entering the sync journal.
public var isValid: Bool {
@@ -28,4 +44,22 @@ public nonisolated struct SFTPPreferences: Codable, Hashable, Sendable {
public var availablePerHostRange: ClosedRange<Int> {
Self.perHostLimitRange.lowerBound...min(Self.perHostLimitRange.upperBound, normalized.globalLimit)
}
private enum CodingKeys: String, CodingKey {
case globalLimit, perHostLimit, strategy
}
public init(from decoder: any Decoder) throws {
let values = try decoder.container(keyedBy: CodingKeys.self)
globalLimit = try values.decodeIfPresent(Int.self, forKey: .globalLimit) ?? Self.defaultGlobalLimit
perHostLimit = try values.decodeIfPresent(Int.self, forKey: .perHostLimit) ?? Self.defaultPerHostLimit
strategy = try values.decodeIfPresent(SFTPConcurrencyStrategy.self, forKey: .strategy) ?? .smart
}
public func encode(to encoder: any Encoder) throws {
var values = encoder.container(keyedBy: CodingKeys.self)
try values.encode(globalLimit, forKey: .globalLimit)
try values.encode(perHostLimit, forKey: .perHostLimit)
try values.encode(strategy, forKey: .strategy)
}
}
@@ -19,11 +19,14 @@ extension SFTPPreferences {
globalLimit: defaults.object(forKey: SettingsKey.sftpGlobalTransferLimit) as? Int
?? Self.defaultGlobalLimit,
perHostLimit: defaults.object(forKey: SettingsKey.sftpPerHostTransferLimit) as? Int
?? Self.defaultPerHostLimit)
?? Self.defaultPerHostLimit,
strategy: defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy)
.flatMap(SFTPConcurrencyStrategy.init(rawValue:)) ?? .smart)
}
public func mirror(to defaults: UserDefaults) {
defaults.set(globalLimit, forKey: SettingsKey.sftpGlobalTransferLimit)
defaults.set(perHostLimit, forKey: SettingsKey.sftpPerHostTransferLimit)
defaults.set(strategy.rawValue, forKey: SettingsKey.sftpConcurrencyStrategy)
}
}
@@ -8,6 +8,17 @@ public struct SFTPSettingsFields: View {
public init(model: SFTPSettingsModel) { self.model = model }
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)
}
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)
}
Stepper(value: binding(\.globalLimit), in: SFTPPreferences.globalLimitRange) {
label("Concurrent Transfers", value: model.preferences.globalLimit)
}
@@ -15,7 +26,7 @@ public struct SFTPSettingsFields: View {
label("Per-Host Transfers", value: model.preferences.perHostLimit)
}
Text(
"Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, the session transfers files one at a time."
"Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, transfers use parallel channels on the existing connection."
)
.font(.caption).foregroundStyle(.secondary)
if let error = model.storageError { Text(error).foregroundStyle(.red) }
@@ -39,4 +50,8 @@ public struct SFTPSettingsFields: View {
private func binding(_ keyPath: WritableKeyPath<SFTPPreferences, Int>) -> Binding<Int> {
Binding(get: { model.preferences[keyPath: keyPath] }, set: { model.set($0, for: keyPath) })
}
private var strategyBinding: Binding<SFTPConcurrencyStrategy> {
Binding(get: { model.preferences.strategy }, set: { model.setStrategy($0) })
}
}
@@ -13,8 +13,10 @@ nonisolated public protocol SFTPToolDefinition: AgentToolDefinition, AgentToolSc
static var extraProperties: [String: AgentJSONValue] { get }
static var extraRequired: [String] { get }
static var presentation: AgentToolDescriptor.Presentation { get }
static var schedulesDataTransfer: Bool { get }
var authentication: AgentSSHAuthenticationRegistry { get }
var connections: AgentConnectionPool { get }
func estimatedByteCount(arguments: Arguments) throws -> Int64?
func perform(
client: any SFTPClientProtocol, path: String, arguments: Arguments,
invocation: AgentToolInvocation
@@ -27,6 +29,9 @@ nonisolated extension SFTPToolDefinition {
}
public static var extraProperties: [String: AgentJSONValue] { [:] }
public static var extraRequired: [String] { [] }
public static var schedulesDataTransfer: Bool { false }
public func estimatedByteCount(arguments _: Arguments) throws -> Int64? { nil }
public var descriptor: AgentToolDescriptor {
var properties: [String: AgentJSONValue] = [
@@ -60,7 +65,9 @@ nonisolated extension SFTPToolDefinition {
let path = try AgentToolSupport.normalizeRemotePath(arguments.string("path"))
return try await AgentToolSupport.withAuthenticatedSFTP(
hostID: hostID, invocation: invocation, context: context,
authentication: authentication, pool: connections
authentication: authentication, pool: connections,
byteCount: try estimatedByteCount(arguments: arguments),
schedulesDataTransfer: Self.schedulesDataTransfer
) { client in
try await perform(
client: client, path: path, arguments: arguments, invocation: invocation
@@ -131,6 +138,7 @@ nonisolated public struct SFTPReadTool: SFTPToolDefinition {
"offset": integer(min: 1, max: 10_000_000), "limit": integer(min: 1, max: 2_000),
]
public static let approvalPolicy: AgentToolDescriptor.ApprovalPolicy = .approve
public static let schedulesDataTransfer = true
public static let presenter = AgentToolDetailPresenter(
id: "builtin.sftp_read", present: present
)
@@ -140,6 +148,7 @@ nonisolated public struct SFTPReadTool: SFTPToolDefinition {
)
public let authentication: AgentSSHAuthenticationRegistry
public let connections: AgentConnectionPool
public func estimatedByteCount(arguments _: Arguments) throws -> Int64? { 50 * 1_024 }
public func perform(
client: any SFTPClientProtocol, path: String, arguments: Arguments,
invocation: AgentToolInvocation
@@ -171,6 +180,7 @@ nonisolated public struct SFTPWriteTool: SFTPToolDefinition {
public static let name = "sftp_write"
public static let summary = "Write a bounded UTF-8 remote file atomically."
public static let approvalPolicy: AgentToolDescriptor.ApprovalPolicy = .ask
public static let schedulesDataTransfer = true
public static let extraProperties = ["content": string(max: 131_072), "overwrite": boolean()]
public static let extraRequired = ["content"]
public static let presenter = AgentToolDetailPresenter(
@@ -184,6 +194,9 @@ nonisolated public struct SFTPWriteTool: SFTPToolDefinition {
)
public let authentication: AgentSSHAuthenticationRegistry
public let connections: AgentConnectionPool
public func estimatedByteCount(arguments: Arguments) throws -> Int64? {
Int64(try arguments.string("content").utf8.count)
}
public func perform(
client: any SFTPClientProtocol, path: String, arguments: Arguments,
invocation: AgentToolInvocation
@@ -620,11 +633,15 @@ nonisolated extension AgentToolSupport {
hostID: Host.ID, invocation: AgentToolInvocation,
context: AgentToolExecutionContext,
authentication: AgentSSHAuthenticationRegistry, pool: AgentConnectionPool,
byteCount: Int64? = nil,
schedulesDataTransfer: Bool = false,
operation: @escaping @Sendable (any SFTPClientProtocol) async throws -> Value
) async throws -> Value {
try await authentication.withSFTP(
hostID: hostID, invocationID: invocation.id,
interaction: context.sshInteraction, pool: pool
interaction: context.sshInteraction, pool: pool,
runID: context.runID, byteCount: byteCount,
schedulesDataTransfer: schedulesDataTransfer
) { client in
do {
return try await operation(client)
@@ -338,12 +338,11 @@ nonisolated public struct SSHExecTool: AgentToolDefinition, AgentToolSchemaBuild
// turn serial, so nothing observes the host while it is
// being changed.
//
// The ceiling on what this buys is worth stating: commands
// aimed at one host queue behind each other inside
// `SSHExecSession` regardless, because a libssh2 session
// cannot be driven from two places at once. The win is
// several hosts at once, and host work overlapping `fetch`,
// `web_search`, and MCP.
// Safe commands may overlap even when aimed at one host:
// `SSHExecSession` gives each command its own Channel on the
// shared transport, while one fair owner serializes the actual
// libssh2 calls. Commands on separate hosts can also run at once,
// as can host work alongside `fetch`, `web_search`, and MCP.
concurrency: safe ? .parallel : .sequential
)
}
@@ -790,42 +790,66 @@ public actor AgentSSHAuthenticationRegistry {
hostID: Host.ID, invocationID: AgentToolInvocation.ID,
interaction: AgentSSHInteraction, pool: AgentConnectionPool,
cancellation: TransferCancellationToken? = nil,
runID: UUID? = nil,
byteCount: Int64? = nil,
schedulesDataTransfer: Bool = false,
concurrencyPermit: SFTPConcurrencyPermit? = nil,
operation: @escaping @Sendable (any SFTPClientProtocol) async throws -> Value
) async throws -> Value {
try await withSession(
hostID: hostID, invocationID: invocationID, interaction: interaction
) { authentication in
let lease = try await self.openAuthenticated(
hostID: hostID, authentication: authentication, pool: pool, kind: .sftp,
evictsPooledConnection: false, cancellation: cancellation
) { attempt in
let resolver: AgentConnectionPool.RequestResolving?
if let attempt {
let request = attempt.request
resolver = { _ in request }
} else {
resolver = nil
}
return try await pool.acquireSFTP(
hostID: hostID, cancellation: cancellation,
configuration: resolver)
}
do {
let value = try await operation(lease.client)
await pool.releaseSFTP(lease.id)
return value
} catch {
await pool.releaseSFTP(lease.id)
if let failure = error as? SFTPError {
switch failure {
case .notConnected, .connectionLost:
await pool.disconnect(hostID: hostID, kind: .sftp)
case .cancelled, .localFile, .operation:
break
let ownedPermit: SFTPConcurrencyPermit?
if schedulesDataTransfer, concurrencyPermit == nil {
ownedPermit = try await pool.sftpConcurrencyScheduler.acquire(
anchorHostID: hostID,
hostIDs: [hostID],
source: .agentRun(runID ?? invocationID),
byteCount: byteCount
)
} else {
ownedPermit = nil
}
let permit = concurrencyPermit ?? ownedPermit
do {
let value = try await withSession(
hostID: hostID, invocationID: invocationID, interaction: interaction
) { authentication in
let lease = try await self.openAuthenticated(
hostID: hostID, authentication: authentication, pool: pool, kind: .sftp,
evictsPooledConnection: false, cancellation: cancellation
) { attempt in
let resolver: AgentConnectionPool.RequestResolving?
if let attempt {
let request = attempt.request
resolver = { _ in request }
} else {
resolver = nil
}
return try await pool.acquireSFTP(
hostID: hostID, cancellation: cancellation,
configuration: resolver,
mode: permit?.schedulingMode ?? .channelsOnly)
}
do {
let value = try await operation(lease.client)
await pool.releaseSFTP(lease.id)
return value
} catch {
await pool.releaseSFTP(lease.id)
if let failure = error as? SFTPError {
switch failure {
case .notConnected, .connectionLost:
await pool.disconnect(hostID: hostID, kind: .sftp)
case .cancelled, .localFile, .operation:
break
}
}
throw error
}
throw error
}
await ownedPermit?.release()
return value
} catch {
await ownedPermit?.release()
throw error
}
}
@@ -907,11 +931,11 @@ public actor AgentSSHAuthenticationRegistry {
/// is what gives it back.
///
/// Exec and SFTP remain separate lifecycle domains. Exec never opens a second
/// transport; overlapping SFTP leases prefer additional connections and fall
/// back to sibling Channels on the primary transport.
/// transport; SFTP data permits resolve their topology from the process-wide
/// preference, while control work and Channel fallback use the primary transport.
public actor AgentConnectionPool {
public nonisolated static let execSchedulingMode = SSHChannelSchedulingMode.channelsOnly
public nonisolated static let sftpSchedulingMode = SSHChannelSchedulingMode.connectionsPreferred
public nonisolated let sftpConcurrencyScheduler: SFTPConcurrencyScheduler
public nonisolated enum Kind: String, Hashable, Sendable, CaseIterable { case exec, sftp }
@@ -1009,7 +1033,8 @@ public actor AgentConnectionPool {
/// SFTP Channel on the primary transport instead.
func acquireSFTP(
hostID: UUID, cancellation: TransferCancellationToken? = nil,
configuration override: RequestResolving? = nil
configuration override: RequestResolving? = nil,
mode: SSHChannelSchedulingMode = .connectionsPreferred
) async throws -> SFTPLease {
try Task.checkCancellation()
if cancellation?.isCancelled == true { throw SFTPError.cancelled }
@@ -1029,7 +1054,7 @@ public actor AgentConnectionPool {
}
if let idle = sftpStates[hostID]?.records.values
.filter({ !$0.busy })
.filter({ !$0.busy && (mode == .connectionsPreferred || $0.id == sftpStates[hostID]?.primaryID) })
.sorted(by: { $0.lastUsedAt < $1.lastUsedAt }).first
{
sftpStates[hostID]?.records[idle.id]?.busy = true
@@ -1042,6 +1067,31 @@ public actor AgentConnectionPool {
throw SFTPError.notConnected
}
if mode == .channelsOnly {
do {
let sibling = try await primaryRecord.client.openSiblingChannel()
if cancellation?.isCancelled == true {
await sibling.disconnect()
throw SFTPError.cancelled
}
return registerSFTPLease(
hostID: hostID, recordID: nil, client: sibling, cancellation: cancellation)
} catch {
try Task.checkCancellation()
if cancellation?.isCancelled == true { throw SFTPError.cancelled }
if error is CancellationError { throw error }
if let failure = error as? SFTPError, case .cancelled = failure { throw failure }
while sftpStates[hostID]?.records[primaryRecord.id]?.busy == true {
try await Task.sleep(for: .milliseconds(10))
try Task.checkCancellation()
if cancellation?.isCancelled == true { throw SFTPError.cancelled }
}
return try await acquireSFTP(
hostID: hostID, cancellation: cancellation,
configuration: override, mode: mode)
}
}
if state.capability == .unknown {
let probe: SFTPProbe
if let existing = sftpProbes[hostID] {
@@ -1068,7 +1118,7 @@ public actor AgentConnectionPool {
}
return try await acquireSFTP(
hostID: hostID, cancellation: cancellation,
configuration: override)
configuration: override, mode: mode)
} catch {
if sftpProbes[hostID]?.id == probe.id {
sftpProbes[hostID] = nil
@@ -1138,6 +1188,7 @@ public actor AgentConnectionPool {
routeFingerprint: @escaping FingerprintResolving,
openExec: @escaping ExecOpening,
openSFTP: @escaping SFTPOpening,
sftpConcurrencyScheduler: SFTPConcurrencyScheduler = SFTPConcurrencyScheduler(),
idleTimeout: Duration = AgentConnectionPool.defaultIdleTimeout,
sweepInterval: Duration? = .seconds(30)
) {
@@ -1145,6 +1196,7 @@ public actor AgentConnectionPool {
self.routeFingerprint = routeFingerprint
self.openExec = openExec
self.openSFTP = openSFTP
self.sftpConcurrencyScheduler = sftpConcurrencyScheduler
self.idleTimeout = idleTimeout
sweeper = AgentIdleSweeper(interval: sweepInterval)
}
@@ -24,7 +24,8 @@ public struct AgentHostServices {
public static func live(
store: HostLibraryModel,
coordinator: RemoteConnectionCoordinator = RemoteConnectionCoordinator()
coordinator: RemoteConnectionCoordinator = RemoteConnectionCoordinator(),
sftpConcurrencyScheduler: SFTPConcurrencyScheduler = SFTPConcurrencyScheduler()
) -> AgentHostServices {
let resolver = AgentHostResolver(store: store)
return AgentHostServices(
@@ -33,7 +34,8 @@ public struct AgentHostServices {
configuration: { try await resolver.configuration(for: $0) },
routeFingerprint: { try await resolver.routeFingerprint(for: $0) },
openExec: { try await coordinator.connectExec(using: $0) },
openSFTP: { try await coordinator.connectSFTP(using: $0) }
openSFTP: { try await coordinator.connectSFTP(using: $0) },
sftpConcurrencyScheduler: sftpConcurrencyScheduler
),
authentication: AgentSSHAuthenticationRegistry(resolver: resolver),
background: AgentBackgroundCommands()
@@ -54,7 +54,8 @@ nonisolated public struct ScratchPullTool: AgentToolDefinition, AgentToolSchemaB
let hostID = try arguments.uuid("host_id")
let data = try await AgentToolSupport.withAuthenticatedSFTP(
hostID: hostID, invocation: invocation, context: context,
authentication: authentication, pool: pool
authentication: authentication, pool: pool,
schedulesDataTransfer: true
) { client in
try await client.readFile(at: remote, maximumBytes: AgentScratchWorkspace.Limits.fileBytes)
}
@@ -137,7 +138,8 @@ nonisolated public struct ScratchPushTool: AgentToolDefinition, AgentToolSchemaB
let hostID = try arguments.uuid("host_id")
try await AgentToolSupport.withAuthenticatedSFTP(
hostID: hostID, invocation: invocation, context: context,
authentication: authentication, pool: pool
authentication: authentication, pool: pool,
byteCount: Int64(file.bytes), schedulesDataTransfer: true
) { client in
try await client.writeFile(
file.data, at: remote, overwrite: arguments.optionalBool("overwrite") ?? false
@@ -96,29 +96,67 @@ nonisolated public struct SFTPCopyTool: AgentToolDefinition, AgentToolSchemaBuil
let request = try Request(invocation)
let cancellation = TransferCancellationToken()
return try await withTaskCancellationHandler {
try await authentication.withSFTP(
hostID: request.sourceHostID, invocationID: invocation.id,
interaction: context.sshInteraction, pool: connections, cancellation: cancellation
) { source in
try await authentication.withSFTP(
hostID: request.destinationHostID, invocationID: invocation.id,
interaction: context.sshInteraction, pool: connections, cancellation: cancellation
) { destination in
let outcome = await SFTPRemoteCopyOperation.run(
sourcePath: request.sourcePath, destinationPath: request.destinationPath,
sameHost: request.sourceHostID == request.destinationHostID,
source: source, destination: destination, policy: request.policy, cancellation: cancellation
) { outcome, progress in
context.report(Self.copyResult(invocation, request, context, outcome, progress: progress))
let permit = try await connections.sftpConcurrencyScheduler.acquire(
anchorHostID: request.destinationHostID,
hostIDs: [request.sourceHostID, request.destinationHostID],
source: .agentRun(context.runID),
byteCount: nil
)
do {
let result = try await authentication.withSFTP(
hostID: request.sourceHostID, invocationID: invocation.id,
interaction: context.sshInteraction, pool: connections, cancellation: cancellation,
runID: context.runID, schedulesDataTransfer: true, concurrencyPermit: permit
) { source in
if request.sourceHostID == request.destinationHostID,
permit.schedulingMode == .channelsOnly
{
return await performCopy(
source: source, destination: source,
request: request, invocation: invocation,
context: context, cancellation: cancellation)
}
return try await authentication.withSFTP(
hostID: request.destinationHostID, invocationID: invocation.id,
interaction: context.sshInteraction, pool: connections, cancellation: cancellation,
runID: context.runID, schedulesDataTransfer: true, concurrencyPermit: permit
) { destination in
await performCopy(
source: source, destination: destination,
request: request, invocation: invocation,
context: context, cancellation: cancellation)
}
return Self.copyResult(invocation, request, context, outcome)
}
await permit.release()
return result
} catch {
await permit.release()
throw error
}
} onCancel: {
cancellation.cancel()
}
}
private func performCopy(
source: any SFTPClientProtocol,
destination: any SFTPClientProtocol,
request: Request,
invocation: AgentToolInvocation,
context: AgentToolExecutionContext,
cancellation: TransferCancellationToken
) async -> AgentToolResult {
let outcome = await SFTPRemoteCopyOperation.run(
sourcePath: request.sourcePath, destinationPath: request.destinationPath,
sameHost: request.sourceHostID == request.destinationHostID,
source: source, destination: destination, policy: request.policy,
cancellation: cancellation
) { outcome, progress in
context.report(Self.copyResult(invocation, request, context, outcome, progress: progress))
}
return Self.copyResult(invocation, request, context, outcome)
}
private static func copyResult(
_ invocation: AgentToolInvocation, _ request: Request, _ context: AgentToolExecutionContext,
_ outcome: SFTPRemoteCopyOperation.Outcome, progress: SFTPProgress? = nil
@@ -0,0 +1,220 @@
import Foundation
import TermindDomain
public nonisolated enum SFTPWorkSource: Hashable, Sendable {
case workspace(UUID)
case agentRun(UUID)
case batch(UUID)
}
public actor SFTPConcurrencyPermit {
public nonisolated let schedulingMode: SSHChannelSchedulingMode
private var releaseAction: (@Sendable () async -> Void)?
fileprivate init(
schedulingMode: SSHChannelSchedulingMode,
release: @escaping @Sendable () async -> Void
) {
self.schedulingMode = schedulingMode
releaseAction = release
}
public func release() async {
let action = releaseAction
releaseAction = nil
await action?()
}
}
/// Process-wide admission for SFTP data work.
///
/// Capacity is counted in logical file operations. A remote copy therefore
/// consumes one global slot and one slot on each distinct endpoint Host even
/// though its implementation may hold two SFTP leases.
public actor SFTPConcurrencyScheduler {
private struct Request {
var id: UUID
var anchorHostID: UUID
var hostIDs: Set<UUID>
var source: SFTPWorkSource
var byteCount: Int64?
var exclusiveResourceIDs: Set<UUID>
var continuation: CheckedContinuation<SFTPConcurrencyPermit, any Error>
}
private struct ActiveRequest {
var hostIDs: Set<UUID>
var exclusiveResourceIDs: Set<UUID>
}
private var preferences: SFTPPreferences
private var requests: [UUID: Request] = [:]
private var queues: [UUID: [SFTPWorkSource: [UUID]]] = [:]
private var hostOrder: [UUID] = []
private var sourceOrder: [UUID: [SFTPWorkSource]] = [:]
private var lastGrantedHost: UUID?
private var lastGrantedSource: [UUID: SFTPWorkSource] = [:]
private var active: [UUID: ActiveRequest] = [:]
private var activePerHost: [UUID: Int] = [:]
private var activeExclusiveResources: Set<UUID> = []
public init(preferences: SFTPPreferences = SFTPPreferences()) {
self.preferences = preferences.normalized
}
public func updatePreferences(_ value: SFTPPreferences) {
preferences = value.normalized
drain()
}
public func acquire(
anchorHostID: UUID,
hostIDs: Set<UUID>,
source: SFTPWorkSource,
byteCount: Int64?,
exclusiveResourceIDs: Set<UUID> = []
) async throws -> SFTPConcurrencyPermit {
try Task.checkCancellation()
let id = UUID()
let normalizedHosts = hostIDs.union([anchorHostID])
return try await withTaskCancellationHandler {
try await withCheckedThrowingContinuation { continuation in
let request = Request(
id: id,
anchorHostID: anchorHostID,
hostIDs: normalizedHosts,
source: source,
byteCount: byteCount,
exclusiveResourceIDs: exclusiveResourceIDs,
continuation: continuation
)
requests[id] = request
if queues[anchorHostID] == nil {
queues[anchorHostID] = [:]
hostOrder.append(anchorHostID)
}
if queues[anchorHostID]?[source] == nil {
queues[anchorHostID]?[source] = []
sourceOrder[anchorHostID, default: []].append(source)
}
queues[anchorHostID]?[source, default: []].append(id)
drain()
}
} onCancel: {
Task { await self.cancel(id) }
}
}
public var activeCount: Int { active.count }
public var queuedCount: Int { requests.count }
public func activeCount(for hostID: UUID) -> Int {
activePerHost[hostID, default: 0]
}
private func cancel(_ id: UUID) {
guard let request = requests.removeValue(forKey: id) else { return }
removeQueuedID(id, request: request)
request.continuation.resume(throwing: CancellationError())
pruneEmptyQueues()
drain()
}
private func release(_ id: UUID) {
guard let request = active.removeValue(forKey: id) else { return }
for hostID in request.hostIDs {
let remaining = max(0, activePerHost[hostID, default: 1] - 1)
activePerHost[hostID] = remaining == 0 ? nil : remaining
}
activeExclusiveResources.subtract(request.exclusiveResourceIDs)
drain()
}
private func drain() {
while active.count < preferences.globalLimit, let request = nextEligibleRequest() {
requests[request.id] = nil
removeQueuedID(request.id, request: request)
let permitID = UUID()
active[permitID] = ActiveRequest(
hostIDs: request.hostIDs,
exclusiveResourceIDs: request.exclusiveResourceIDs)
for hostID in request.hostIDs { activePerHost[hostID, default: 0] += 1 }
activeExclusiveResources.formUnion(request.exclusiveResourceIDs)
lastGrantedHost = request.anchorHostID
lastGrantedSource[request.anchorHostID] = request.source
let mode = schedulingMode(byteCount: request.byteCount)
request.continuation.resume(
returning: SFTPConcurrencyPermit(schedulingMode: mode) { [weak self] in
await self?.release(permitID)
})
}
pruneEmptyQueues()
}
private func nextEligibleRequest() -> Request? {
for hostID in rotated(hostOrder, after: lastGrantedHost) {
guard let hostQueues = queues[hostID] else { continue }
let sources = sourceOrder[hostID] ?? []
for source in rotated(sources, after: lastGrantedSource[hostID]) {
guard let id = hostQueues[source]?.first, let request = requests[id] else { continue }
guard request.hostIDs.allSatisfy({ activePerHost[$0, default: 0] < preferences.perHostLimit })
else { continue }
guard request.exclusiveResourceIDs.isDisjoint(with: activeExclusiveResources) else { continue }
return request
}
}
return nil
}
private func schedulingMode(byteCount: Int64?) -> SSHChannelSchedulingMode {
switch preferences.strategy {
case .channelsOnly:
.channelsOnly
case .connectionsPreferred:
.connectionsPreferred
case .smart:
if let byteCount, byteCount > SFTPConcurrencyStrategy.smartConnectionThreshold {
.connectionsPreferred
} else {
.channelsOnly
}
}
}
private func removeQueuedID(_ id: UUID, request: Request) {
guard var values = queues[request.anchorHostID]?[request.source] else { return }
values.removeAll { $0 == id }
queues[request.anchorHostID]?[request.source] = values
}
private func pruneEmptyQueues() {
for hostID in hostOrder {
guard var hostQueues = queues[hostID] else { continue }
for source in sourceOrder[hostID] ?? [] where hostQueues[source]?.isEmpty != false {
hostQueues[source] = nil
sourceOrder[hostID]?.removeAll { $0 == source }
}
if hostQueues.isEmpty {
queues[hostID] = nil
sourceOrder[hostID] = nil
hostOrder.removeAll { $0 == hostID }
} else {
queues[hostID] = hostQueues
}
}
let activeHosts = Set(active.values.flatMap(\.hostIDs))
for hostID in Array(lastGrantedSource.keys)
where queues[hostID] == nil && !activeHosts.contains(hostID) {
lastGrantedSource[hostID] = nil
}
if let lastGrantedHost, queues[lastGrantedHost] == nil, !activeHosts.contains(lastGrantedHost) {
self.lastGrantedHost = nil
}
}
private func rotated<Value: Equatable>(_ values: [Value], after last: Value?) -> [Value] {
guard !values.isEmpty, let last, let index = values.firstIndex(of: last) else { return values }
let split = values.index(after: index)
return Array(values[split...]) + Array(values[..<split])
}
}
@@ -105,20 +105,26 @@ public final class SFTPTransferCoordinator {
private let store: HostLibraryModel
private let clientFactory: ClientFactory
public let concurrencyScheduler: SFTPConcurrencyScheduler
private var runtimes: [SFTPTransfer.ID: Runtime] = [:]
private var remoteCopyRuntimes: [SFTPTransfer.ID: RemoteCopyRuntime] = [:]
private var admissionTasks: [SFTPTransfer.ID: Task<Void, Never>] = [:]
private var preferencesUpdateTask: Task<Void, Never>?
private var activeTasks: [SFTPTransfer.ID: Task<Void, Never>] = [:]
private var activePerHost: [Host.ID: Int] = [:]
private var activePermits: [SFTPTransfer.ID: SFTPConcurrencyPermit] = [:]
private var activeBorrowedConnections: Set<UUID> = []
private var remoteCopySources: [UUID: (source: SFTPRemoteCopySource, createdAt: Date)] = [:]
public private(set) var globalLimit: Int
public private(set) var perHostLimit: Int
public private(set) var strategy: SFTPConcurrencyStrategy
public init(
store: HostLibraryModel,
clientFactory: @escaping ClientFactory,
globalLimit: Int? = nil, perHostLimit: Int? = nil
globalLimit: Int? = nil, perHostLimit: Int? = nil,
strategy: SFTPConcurrencyStrategy? = nil,
concurrencyScheduler: SFTPConcurrencyScheduler? = nil
) {
self.store = store
self.clientFactory = clientFactory
@@ -127,15 +133,30 @@ public final class SFTPTransferCoordinator {
globalLimit: globalLimit ?? defaults.object(forKey: SettingsKey.sftpGlobalTransferLimit) as? Int
?? SFTPPreferences.defaultGlobalLimit,
perHostLimit: perHostLimit ?? defaults.object(forKey: SettingsKey.sftpPerHostTransferLimit) as? Int
?? SFTPPreferences.defaultPerHostLimit)
?? SFTPPreferences.defaultPerHostLimit,
strategy: strategy ?? defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy)
.flatMap(SFTPConcurrencyStrategy.init(rawValue:)) ?? .smart)
self.globalLimit = limits.globalLimit
self.perHostLimit = limits.perHostLimit
self.strategy = limits.strategy
self.concurrencyScheduler = concurrencyScheduler ?? SFTPConcurrencyScheduler(preferences: limits)
}
public func updateLimits(global: Int, perHost: Int) {
let limits = SFTPPreferences(globalLimit: global, perHostLimit: perHost)
updatePreferences(
SFTPPreferences(
globalLimit: global, perHostLimit: perHost,
strategy: strategy
))
}
public func updatePreferences(_ value: SFTPPreferences) {
let limits = value.normalized
globalLimit = limits.globalLimit
perHostLimit = limits.perHostLimit
strategy = limits.strategy
preferencesUpdateTask?.cancel()
preferencesUpdateTask = Task { await concurrencyScheduler.updatePreferences(limits) }
schedule()
}
@@ -330,6 +351,7 @@ public final class SFTPTransferCoordinator {
guard canPause(id), let runtime = runtimes[id] else { return }
runtimes[id]?.pauseRequested = true
if activeTasks[id] == nil {
admissionTasks.removeValue(forKey: id)?.cancel()
store.setSFTPTransferState(.paused, for: id)
runtime.connection.managedSession?.suspendTransfer(id)
} else {
@@ -355,6 +377,7 @@ public final class SFTPTransferCoordinator {
runtimes[id]?.pauseRequested = false
runtime.cancellation.cancel()
if activeTasks[id] == nil {
admissionTasks.removeValue(forKey: id)?.cancel()
let task = Task { [self] in
var state: SFTPTransfer.State = .cancelled
do { try await cleanup(runtime, direction: transfer.direction, client: nil) } catch {
@@ -370,6 +393,7 @@ public final class SFTPTransferCoordinator {
activeTasks[id] = task
}
} else if transfer.state == .queued, activeTasks[id] == nil {
admissionTasks.removeValue(forKey: id)?.cancel()
let sessions = managedSessions(for: id)
guard let runtime = remoteCopyRuntimes.removeValue(forKey: id) else { return }
activeTasks[id] = Task { [self] in
@@ -418,8 +442,8 @@ public final class SFTPTransferCoordinator {
}
public func waitForCancellation() async {
while !activeTasks.isEmpty {
let tasks = Array(activeTasks.values)
while !activeTasks.isEmpty || !admissionTasks.isEmpty {
let tasks = Array(activeTasks.values) + Array(admissionTasks.values)
for task in tasks { await task.value }
}
}
@@ -458,36 +482,56 @@ public final class SFTPTransferCoordinator {
public func clearFinished() { store.clearFinishedSFTPTransfers() }
private func schedule() {
while activeTasks.count < globalLimit {
guard
let sftpTransfer = store.sftpTransfers.first(where: { sftpTransfer in
guard sftpTransfer.state == .queued, activeTasks[sftpTransfer.id] == nil,
for sftpTransfer in store.sftpTransfers where sftpTransfer.state == .queued {
guard activeTasks[sftpTransfer.id] == nil, admissionTasks[sftpTransfer.id] == nil,
runtimes[sftpTransfer.id]?.cancelRequested != true,
runtimes[sftpTransfer.id] != nil || remoteCopyRuntimes[sftpTransfer.id] != nil
else { continue }
let hostIDs = Set(hostIDs(for: sftpTransfer))
guard let anchorHostID = anchorHostID(for: sftpTransfer), !hostIDs.isEmpty else { continue }
let source = workSource(for: sftpTransfer)
let exclusiveResourceIDs = Set(borrowedIDs(for: sftpTransfer))
let scheduler = concurrencyScheduler
let preferencesUpdateTask = preferencesUpdateTask
admissionTasks[sftpTransfer.id] = Task { @MainActor [weak self] in
do {
await preferencesUpdateTask?.value
let permit = try await scheduler.acquire(
anchorHostID: anchorHostID,
hostIDs: hostIDs,
source: source,
byteCount: sftpTransfer.byteCount >= 0 ? sftpTransfer.byteCount : nil,
exclusiveResourceIDs: exclusiveResourceIDs
)
guard let self else {
await permit.release()
return
}
admissionTasks[sftpTransfer.id] = nil
guard store.sftpTransfer(sftpTransfer.id)?.state == .queued,
runtimes[sftpTransfer.id]?.cancelRequested != true,
runtimes[sftpTransfer.id] != nil || remoteCopyRuntimes[sftpTransfer.id] != nil
else { return false }
guard managedSessions(for: sftpTransfer.id).allSatisfy(\.canStartTransfer) else { return false }
let borrowedIDs = borrowedIDs(for: sftpTransfer)
guard borrowedIDs.allSatisfy({ !activeBorrowedConnections.contains($0) }) else {
return false
else {
await permit.release()
return
}
return hostIDs(for: sftpTransfer).allSatisfy {
activePerHost[$0, default: 0] < perHostLimit
}
})
else { return }
start(sftpTransfer)
activePermits[sftpTransfer.id] = permit
start(sftpTransfer, mode: permit.schedulingMode)
} catch {
self?.admissionTasks[sftpTransfer.id] = nil
}
}
}
}
private func start(_ sftpTransfer: SFTPTransfer) {
private func start(_ sftpTransfer: SFTPTransfer, mode: SSHChannelSchedulingMode) {
for session in managedSessions(for: sftpTransfer.id) { session.reserveSlot() }
if let runtime = remoteCopyRuntimes[sftpTransfer.id] {
startRemoteCopy(sftpTransfer, runtime: runtime)
startRemoteCopy(sftpTransfer, runtime: runtime, mode: mode)
return
}
guard let hostID = sftpTransfer.hostID, let runtime = runtimes[sftpTransfer.id] else { return }
store.setSFTPTransferState(.running, for: sftpTransfer.id)
activePerHost[hostID, default: 0] += 1
let borrowedIDs = Array(runtime.connection.borrowedID.map { [$0] } ?? [])
activeBorrowedConnections.formUnion(borrowedIDs)
let task = Task { [weak self] in
@@ -497,7 +541,8 @@ public final class SFTPTransferCoordinator {
var lease: ClientLease?
var result: SFTPTransfer.State
do {
let acquired = try await clientLease(for: runtime.connection, cancellation: runtime.cancellation)
let acquired = try await clientLease(
for: runtime.connection, mode: mode, cancellation: runtime.cancellation)
lease = acquired
let client = acquired.client
let progress: @Sendable (SFTPProgress) -> Void = { [weak self] value in
@@ -572,10 +617,13 @@ public final class SFTPTransferCoordinator {
}
}
private func startRemoteCopy(_ sftpTransfer: SFTPTransfer, runtime: RemoteCopyRuntime) {
private func startRemoteCopy(
_ sftpTransfer: SFTPTransfer,
runtime: RemoteCopyRuntime,
mode: SSHChannelSchedulingMode
) {
let involvedHostIDs = Array(Set([runtime.sourceHostID, runtime.destinationHostID]))
store.setSFTPTransferState(.running, for: sftpTransfer.id)
for hostID in involvedHostIDs { activePerHost[hostID, default: 0] += 1 }
let borrowedIDs = [
runtime.sourceConnection.borrowedID,
runtime.destinationConnection.borrowedID,
@@ -591,18 +639,18 @@ public final class SFTPTransferCoordinator {
let destinationID = runtime.destinationConnection.managedSession?.id
if let sourceID, sourceID == destinationID {
sourceLease = try await clientLease(
for: runtime.sourceConnection, cancellation: runtime.cancellation)
for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation)
destinationLease = sourceLease
} else if let sourceID, let destinationID, sourceID.uuidString > destinationID.uuidString {
destinationLease = try await clientLease(
for: runtime.destinationConnection, cancellation: runtime.cancellation)
for: runtime.destinationConnection, mode: mode, cancellation: runtime.cancellation)
sourceLease = try await clientLease(
for: runtime.sourceConnection, cancellation: runtime.cancellation)
for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation)
} else {
sourceLease = try await clientLease(
for: runtime.sourceConnection, cancellation: runtime.cancellation)
for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation)
destinationLease = try await clientLease(
for: runtime.destinationConnection, cancellation: runtime.cancellation)
for: runtime.destinationConnection, mode: mode, cancellation: runtime.cancellation)
}
guard let acquiredSource = sourceLease, let acquiredDestination = destinationLease else {
throw SFTPError.notConnected
@@ -650,12 +698,30 @@ public final class SFTPTransferCoordinator {
return sftpTransfer.hostID.map { [$0] } ?? []
}
private func anchorHostID(for sftpTransfer: SFTPTransfer) -> Host.ID? {
sftpTransfer.destinationHostID ?? sftpTransfer.sourceHostID ?? sftpTransfer.hostID
}
private func workSource(for sftpTransfer: SFTPTransfer) -> SFTPWorkSource {
if let runtime = remoteCopyRuntimes[sftpTransfer.id],
let session = runtime.destinationConnection.managedSession
{
return .workspace(session.id)
}
if let session = runtimes[sftpTransfer.id]?.connection.managedSession {
return .workspace(session.id)
}
return .batch(sftpTransfer.batchID)
}
private func clientLease(
for connection: SFTPTransferConnection, cancellation: TransferCancellationToken? = nil
for connection: SFTPTransferConnection,
mode: SSHChannelSchedulingMode = .channelsOnly,
cancellation: TransferCancellationToken? = nil
) async throws -> ClientLease {
switch connection {
case .session(let session):
return try await session.acquire(cancellation: cancellation)
return try await session.acquire(mode: mode, cancellation: cancellation)
case .saved(let configuration):
let client = clientFactory()
do {
@@ -735,20 +801,19 @@ public final class SFTPTransferCoordinator {
session.releaseSlot()
await session.releaseTransfer(id)
}
activeTasks.removeValue(forKey: id)
for hostID in hostIDs {
activePerHost[hostID] = max(0, activePerHost[hostID, default: 1] - 1)
}
releaseBorrowedConnections(borrowedIDs)
if state == .paused {
store.setSFTPTransferState(.paused, for: id)
schedule()
return
} else {
let completion =
runtimes.removeValue(forKey: id)?.completion
?? remoteCopyRuntimes.removeValue(forKey: id)?.completion
store.finishSFTPTransfer(id, state: state)
completion?(state)
}
let completion =
runtimes.removeValue(forKey: id)?.completion
?? remoteCopyRuntimes.removeValue(forKey: id)?.completion
store.finishSFTPTransfer(id, state: state)
completion?(state)
activeTasks.removeValue(forKey: id)
let permit = activePermits.removeValue(forKey: id)
releaseBorrowedConnections(borrowedIDs)
await permit?.release()
schedule()
}
}
@@ -2,11 +2,10 @@ import Foundation
import TermindApplication
/// A browser and its queued transfers share ownership of one authenticated connection.
/// Extra connections are always opened without interaction, using this generation's snapshot.
/// The admitted topology decides between sibling Channels and additional connections;
/// additional connections are always opened without interaction from this generation's snapshot.
@MainActor
public final class SFTPTransferSession {
public nonisolated static let schedulingMode = SSHChannelSchedulingMode.connectionsPreferred
public let id = UUID()
public let configuration: SSHConnectionConfiguration
private var clientFactory: SFTPTransferCoordinator.ClientFactory
@@ -81,10 +80,30 @@ public final class SFTPTransferSession {
await old?.disconnect()
}
func acquire(cancellation: TransferCancellationToken? = nil) async throws -> SFTPTransferLease {
func acquire(
mode: SSHChannelSchedulingMode,
cancellation: TransferCancellationToken? = nil
) async throws -> SFTPTransferLease {
try checkCancellation(cancellation)
while probing { try await waitForPrimary(cancellation: cancellation) }
if capability != .channels, primary != nil, !(prefersPrimary && !primaryBusy) {
if mode == .channelsOnly, let primary, !(prefersPrimary && !primaryBusy) {
do {
let client = try await primary.openSiblingChannel()
try checkCancellation(cancellation)
return SFTPTransferLease(client: client) { await client.disconnect() }
} catch {
try checkCancellation(cancellation)
if error is CancellationError { throw error }
if let failure = error as? SFTPError, case .cancelled = failure { throw failure }
// A server may accept only the browser's original SFTP
// subsystem. Keep the Channel-only transport contract and
// serialize this lease on that primary rather than opening a
// second SSH connection or pausing the transfer.
}
}
if mode == .connectionsPreferred, capability != .channels, primary != nil,
!(prefersPrimary && !primaryBusy)
{
let isProbe = capability == .unknown
if isProbe { probing = true }
defer {
@@ -106,12 +125,18 @@ public final class SFTPTransferSession {
availabilityChanged?()
}
}
if capability == .channels, let primary {
try checkCancellation(cancellation)
let client = try await primary.openSiblingChannel()
try checkCancellation(cancellation)
availabilityChanged?()
return SFTPTransferLease(client: client) { await client.disconnect() }
if mode == .connectionsPreferred, capability == .channels, let primary {
do {
try checkCancellation(cancellation)
let client = try await primary.openSiblingChannel()
try checkCancellation(cancellation)
availabilityChanged?()
return SFTPTransferLease(client: client) { await client.disconnect() }
} catch {
try checkCancellation(cancellation)
if error is CancellationError { throw error }
if let failure = error as? SFTPError, case .cancelled = failure { throw failure }
}
}
while primaryBusy {
try await waitForPrimary(cancellation: cancellation)
+22 -14
View File
@@ -164,19 +164,24 @@ AgentKit is a separately versioned remote package; its sibling working tree is n
- Domain owns `RemoteFile`, Application `RemoteFileSystem`/`MobileSFTPSessionModel`, Remote Infrastructure libssh2 protocols,
Presentation symbols/kinds. Device-local History changes only after fresh successful listings/downloads, never stale/failed/
cancelled/type-changed work.
- Generation-scoped `SFTPTransferSession` outlives panes. First additional connection silently probes the full route: success
enables connection-parallel workers; failure opens sibling SFTP Channels on the primary transport, never the browser's control
Channel. Later failure degrades pending work without replaying active operations. Cancellation caches no failure. Terminal
preflight donates its successful connection. Refresh durable trust, never promote once-only trust.
- App queue defaults 15 active files/five per Host.ID across sessions; ranges 1–100 and 1–min(50, global). The unit is an active
SFTP lease: an independent connection and a sibling Channel consume the same slot. `SFTPSettingsModel` scheduling outlives
Settings. New limits affect future work only. Pane close releases only its ownership; queued work retains the primary
connection, paused work configuration/checkpoints but no connection. Retired generations cannot replace snapshots.
- Generation-scoped `SFTPTransferSession` outlives panes. SFTP data work resolves one topology per admitted file: Channels Only
opens sibling SFTP Channels, Connections Preferred silently probes the full route and falls back to siblings, and Smart uses
Channels through 64 MB/64,000,000 bytes (including unknown sizes) before preferring connections. A failed silent probe degrades later work
without replaying active operations; cancellation caches no failure. Browser control Channels stay independent, terminal
preflight may donate its successful connection, and durable trust refresh never promotes once-only trust.
- One process scheduler covers Workspace/terminal/edit transfers and Agent data operations. It defaults to 30 active files/ten
per Host.ID (ranges 1–100 and 1–min(50, global)), rotating Hosts then Workspace/Agent-run/batch sources while retaining FIFO
within a source. A Channel and an independent connection consume the same logical-file capacity. Cross-Host copy consumes one
global slot and one slot on each endpoint Host. Browser and Agent metadata/control operations do not consume data slots.
`SFTPSettingsModel` scheduling outlives Settings; new limits/topology affect queued work only. Pane close releases only its
ownership; queued work retains the primary connection, paused work configuration/checkpoints but no connection.
- Checkpoints, references, acknowledged offsets and digests are memory-only. Pause/loss retains partial destinations; explicit
resume verifies source/prefix. No ordinary staging/bookmarks/restart recovery or silent replay. Cancel/Host delete removes
partials and reports failed cleanup. Mobile expiration pauses upload/download, cancels copies; desktop quit allows three seconds.
Same-connection copies alternate bounded reads/writes, validate size/metadata/EOF and atomically commit temporary files.
- Shared planners retain directory-only batch destinations/mutation counts. Apps own UI, cancellation/errors/refresh and deletion.
- Shared planners retain directory-only batch destinations/mutation counts. In-app SFTP-to-SFTP drag/copy consumes the private
remote payload before asking for its Finder file promise, streams directly between clients without local staging, and exposes
exactly one `.copy` queue row per regular file. Apps own UI, cancellation/errors/refresh and deletion.
`SFTPRemoteConflictResolver` validates recursive child paths, subtree mapping/Keep Both: upload blocks strict descendants; copy
also blocks conflicting root. Desktop context menus snapshot reconciled selection; appearance never mutates it.
- Terminal drop preparation is cancellable/generation-scoped; late results cannot enqueue. Nonmodal notices cannot replace same-batch
@@ -197,10 +202,12 @@ AgentKit is a separately versioned remote package; its sibling working tree is n
### Agent/MCP byte channels
- Agent SSH keeps one Host transport and multiplexes concurrent exec Channels. Agent SFTP uses connection-preferred scheduling
with sibling-Channel fallback and a 180-second idle pool across conversations/MCP. `sftp_copy` belongs to Agent SFTP: regular
trees to existing directories, browser conflict/merge policy, preserved sources and two ordered endpoint leases; same-Host
copies prefer two connections and fall back to two Channels. Active exec/SFTP leases block idle sweeping; background commands
- Agent SSH keeps one Host transport and multiplexes concurrent exec Channels. Agent SFTP data work shares the process SFTP
scheduler and resolves Smart/Channel-only/connection-preferred topology by the synced preference; metadata/control calls use
Channels without consuming data capacity. Additional Agent connections retain a 180-second idle pool across conversations/MCP.
`sftp_copy` belongs to Agent SFTP: regular trees to existing directories, browser conflict/merge policy and preserved sources.
Cross-Host copies take two ordered endpoint leases; same-Host copies use two leases for Connections Preferred and one SFTP
Channel for Channel-only/Smart-unknown work. Active exec/SFTP leases block idle sweeping; background commands
retain only their Channel, not the whole transport. Started operations never replay. No model bytes/staging/scratch.
- macOS MCP exposes the same definition; Bearer/tool switches/Full access remain boundaries, auth uses external interaction. Stop awaits
copies. Audit metadata/counts/endpoints/policy, never bodies. Stdio only forwards protocol; mobile has no local MCP listener.
@@ -348,7 +355,8 @@ factories/Sendable buffer handlers. Timers/workers enter explicit short MainActo
- Frozen V13 Settings; V14 terminal, V15 Metrics journal/account/quarantine, V21 SFTP. Commit data+intent atomically, then mirror defaults; remote
apply refreshes without echo. Separate records cover providers/models/MCP/skills/roles/search/terminal/custom text/SFTP limits. First SFTP row
imports prior local values once, missing 15/5; future preparation uses persisted truth. `v1/sftp/default` uses `SFTPPreferencesItem` in same zone/
V1 envelope so old clients ignore it before decode. Invalid limits/type mismatches quarantine. Running Agent snapshots remain fixed.
V1 envelope so old clients ignore it before decode. Missing strategy decodes as Smart; new/missing records default to 30/10 while
existing saved limits remain unchanged. Invalid limits/type mismatches quarantine. Running Agent snapshots remain fixed.
- Live role records sync explicit choices, not resolved fallbacks: nil generator = Same as Chat, nil Guardian = Automatic. Read legacy
`securityReview`, migrate to canonical `guardian` without reusing system fields. Conversations/approval mode/last-used model/listener config/
unrelated defaults stay local. Listener token is this-device-only Data Protection Keychain.
+19
View File
@@ -175,3 +175,22 @@ never justifies extraction. Keep schemas, native lists/terminal teardown and loc
reproducible `LaunchFileLogServiceTests.loggerDropsUnapprovedSensitiveValues` failure; one concurrent AgentRuntime
cancellation failure passed immediately when its 20-test suite was rerun. Recursive format also retains the unrelated
existing `LoggingFilesSettings.swift:110` finding.
### 2026-09-19 — fair SFTP admission and selectable topology (complete)
- **Evidence:** the app transfer coordinator admitted the first eligible queued row, so one Workspace could fill every
Host slot, while Agent SFTP leases bypassed the 15/5 limits entirely. User and Agent pools also fixed their topology
at connection-preferred even when a small transfer would not repay another SSH handshake. Internal SFTP drag providers
advertised both a private remote payload and a Finder file promise; reading both representations could start a redundant
download beside the direct remote copy.
- **Action:** add one process scheduler shared by Workspace/terminal/edit and Agent data work, rotating Hosts and then stable
Workspace/Agent-run/batch sources. Resolve Smart (64 MB decimal), Channels Only and Connections Preferred only after admission;
retain feature-owned connection lifetimes and connection-to-Channel fallback. Extend the existing synced SFTP preference
record compatibly and consume the private drag payload without materializing its local-file promise.
- **Result:** new/missing settings default to Smart and 30/10 while saved values remain unchanged. Logical file operations
consume identical capacity whether backed by a Channel or connection; cross-Host copy counts once globally and once per
endpoint. Agent control operations remain outside data capacity, `sftp_copy` uses one composite permit, and in-app remote
copy exposes one `.copy` row per regular file without local staging. This is additive scheduling work, not a reduction claim.
- **Verification:** macOS and generic iOS Simulator builds passed. Focused suites passed for scheduler/session topology,
settings persistence/sync, Agent authentication/MCP, Agent/Workspace copy and native drag payload selection. Chinese
localization and diff checks passed; final full-suite/format status is recorded in the task handoff.
+10 -3
View File
@@ -178,12 +178,16 @@ final class MobileAppModel {
service: AppleSpeechTranscriptionService(audioInput: speechAudio)
)
speechAudio.onInterruption = { [weak speech] in speech?.cancelAll() }
let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences)
assistant = MobileAssistantModel(
configuration: ai,
repository: MemoryIndexedAgentRepository(
base: SwiftDataAgentRunRepository(context: ModelContext(container)), memory: memory),
memory: memory,
hostServices: AgentHostServices.live(store: library, coordinator: connectionCoordinator),
hostServices: AgentHostServices.live(
store: library,
coordinator: connectionCoordinator,
sftpConcurrencyScheduler: sftpConcurrencyScheduler),
metrics: metrics.core, skills: skills,
authorize: { [pro] in pro.require(.assistant) },
speech: speech,
@@ -197,7 +201,10 @@ final class MobileAppModel {
forwardings = dependencies.makeForwardings(library)
transfers = SFTPTransferCoordinator(
store: library, clientFactory: { connectionCoordinator.makeSFTPClient() },
globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit)
globalLimit: sftpSettings.preferences.globalLimit,
perHostLimit: sftpSettings.preferences.perHostLimit,
strategy: sftpSettings.preferences.strategy,
concurrencyScheduler: sftpConcurrencyScheduler)
sftpEdits = SFTPEditService(transfers: transfers, host: { [weak library] in library?.host($0) })
library.willDeleteHost = { [weak sftpEdits, weak forwardings] hostID, _ in
sftpEdits?.invalidate(hostID: hostID)
@@ -232,7 +239,7 @@ final class MobileAppModel {
}
vaultSync?.didApplyRemote = { [weak self] in self?.applyRemoteVault($0, keys: $1) }
sftpSettings.didChange = { [weak transfers] preferences in
transfers?.updateLimits(global: preferences.globalLimit, perHost: preferences.perHostLimit)
transfers?.updatePreferences(preferences)
}
settingsSync?.didApplyRemote = { [weak self] snapshot, _ in
self?.ai.applyRemoteSnapshot(snapshot.configuration, webSearch: snapshot.webSearch)
+9 -4
View File
@@ -13,16 +13,21 @@ import Testing
struct MobileSettingsPortTests {
@Test func sftpSettingsUseTheSharedPreferencesAndUpdateWithoutOpeningSettings() throws {
let app = try MobileAppModel(inMemory: true)
#expect(app.sftpSettings.preferences == SFTPPreferences(globalLimit: 15, perHostLimit: 5))
#expect(app.transfers.globalLimit == 15)
#expect(app.transfers.perHostLimit == 5)
#expect(app.sftpSettings.preferences == SFTPPreferences(globalLimit: 30, perHostLimit: 10))
#expect(app.transfers.globalLimit == 30)
#expect(app.transfers.perHostLimit == 10)
#expect(app.transfers.strategy == .smart)
#expect(app.sftpSettings.set(100, for: \.globalLimit))
#expect(app.sftpSettings.set(50, for: \.perHostLimit))
#expect(app.sftpSettings.setStrategy(.connectionsPreferred))
#expect(app.transfers.globalLimit == 100)
#expect(app.transfers.perHostLimit == 50)
app.sftpSettings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2))
#expect(app.transfers.strategy == .connectionsPreferred)
app.sftpSettings.applyRemoteSnapshot(
SFTPPreferences(globalLimit: 3, perHostLimit: 2, strategy: .channelsOnly))
#expect(app.transfers.globalLimit == 3)
#expect(app.transfers.perHostLimit == 2)
#expect(app.transfers.strategy == .channelsOnly)
}
@Test func sftpCoordinatorAppliesNormalizedSettingsImmediately() {
+10 -3
View File
@@ -168,7 +168,11 @@ final class AppModel {
serialPorts = dependencies.serialPortService
agentRunRepository = MemoryIndexedAgentRepository(
base: dependencies.agentRunRepository ?? InMemoryAgentRunRepository(), memory: memory)
let agentHostServices = AgentHostServices.live(store: hosts, coordinator: connectionCoordinator)
let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences)
let agentHostServices = AgentHostServices.live(
store: hosts,
coordinator: connectionCoordinator,
sftpConcurrencyScheduler: sftpConcurrencyScheduler)
agentHosts = agentHostServices.resolver
agentAuthentication = agentHostServices.authentication
agentConnections = agentHostServices.connections
@@ -182,7 +186,10 @@ final class AppModel {
sftpTransfers = SFTPTransferCoordinator(
store: hosts,
clientFactory: { connectionCoordinator.makeSFTPClient() },
globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit
globalLimit: sftpSettings.preferences.globalLimit,
perHostLimit: sftpSettings.preferences.perHostLimit,
strategy: sftpSettings.preferences.strategy,
concurrencyScheduler: sftpConcurrencyScheduler
)
sftpEdits = SFTPEditService(transfers: sftpTransfers, host: { [weak hosts] in hosts?.host($0) })
self.ai = ai
@@ -263,7 +270,7 @@ final class AppModel {
self?.applyRemoteVaultSnapshot(snapshot, changedKeys: keys)
}
sftpSettings.didChange = { [weak sftpTransfers] preferences in
sftpTransfers?.updateLimits(global: preferences.globalLimit, perHost: preferences.perHostLimit)
sftpTransfers?.updatePreferences(preferences)
}
settingsSync?.didApplyRemote = { [weak self] snapshot, _ in
self?.ai.applyRemoteSnapshot(snapshot.configuration, webSearch: snapshot.webSearch)
@@ -49,7 +49,9 @@ final class MCPServerTools {
let handler = await externalAuthentication.handler(clientName: client, runID: runID)
return try await authentication.withSFTP(
hostID: request.hostID, invocationID: invocationID, interaction: handler.sshInteraction,
pool: connections, cancellation: token
pool: connections, cancellation: token,
runID: runID, byteCount: request.byteCount,
schedulesDataTransfer: true
) { client in
try await SFTPStreamUploadOperation.run(
client: client, remotePath: request.remotePath, byteCount: request.byteCount,
@@ -63,7 +65,8 @@ final class MCPServerTools {
let handler = await externalAuthentication.handler(clientName: clientName, runID: runID)
try await authentication.withSFTP(
hostID: request.hostID, invocationID: invocationID, interaction: handler.sshInteraction,
pool: connections, cancellation: token
pool: connections, cancellation: token,
runID: runID, schedulesDataTransfer: true
) { client in
let stream = try await client.openDownloadStream(
at: request.remotePath, maximumBytes: Int64(MCPTransferBudget.maximumBytes), cancellation: token
+18 -3
View File
@@ -223,13 +223,28 @@ nonisolated struct NativeDropPayload: Sendable {
/// `isEnabled` for why refusing the hit test is not enough on its own.
func supportedPayload(from pasteboard: NSPasteboard) -> NativeDropPayload? {
guard isEnabled else { return nil }
let payload = Self.payload(from: pasteboard).restricted(to: acceptedKinds)
let payload = Self.payload(
from: pasteboard,
preferSFTPItems: acceptedKinds.contains(.sftpItems)
).restricted(to: acceptedKinds)
return payload.preferredKind == nil ? nil : payload
}
static func payload(from pasteboard: NSPasteboard) -> NativeDropPayload {
static func payload(
from pasteboard: NSPasteboard,
preferSFTPItems: Bool = true
) -> NativeDropPayload {
let sftpType = NSPasteboard.PasteboardType(UTType.termindSFTPItems.identifier)
var payload = NativeDropPayload(sftpItems: pasteboard.data(forType: sftpType))
let sftpItems = pasteboard.data(forType: sftpType)
if preferSFTPItems, let sftpItems {
// The same provider also advertises a promised local file for
// Finder. Asking that representation for a URL materializes
// the promise and starts a download. An in-app SFTP target
// must consume only the private payload, otherwise one drag
// creates both a remote copy and a redundant download/upload.
return NativeDropPayload(sftpItems: sftpItems)
}
var payload = NativeDropPayload(sftpItems: sftpItems)
for item in pasteboard.pasteboardItems ?? [] {
if let value = item.string(forType: .fileURL),
+25
View File
@@ -98,6 +98,31 @@ import TermindRemoteInfrastructure
#expect(target.supportedPayload(from: pasteboard)?.preferredKind == .sftpItems)
}
@Test
func internalSFTPPayloadDoesNotRequestItsPromisedLocalFile() throws {
let pasteboard = stagedPasteboard()
defer { pasteboard.releaseGlobally() }
let item = NSPasteboardItem()
let remote = try JSONEncoder().encode(SFTPRemoteItemsPayload(token: UUID()))
item.setData(
remote,
forType: NSPasteboard.PasteboardType(UTType.termindSFTPItems.identifier))
item.setString(
URL(fileURLWithPath: "/tmp/materialized-by-a-file-promise").absoluteString,
forType: .fileURL)
pasteboard.writeObjects([item])
let payload = FileURLDropNSView.payload(from: pasteboard)
#expect(payload.sftpItems == remote)
#expect(payload.fileURLs.isEmpty)
#expect(payload.preferredKind == .sftpItems)
let terminalPayload = FileURLDropNSView.payload(
from: pasteboard, preferSFTPItems: false)
#expect(terminalPayload.fileURLs.count == 1)
}
@Test
func finderImagePrefersItsFileURLAndFilename() throws {
let pasteboard = stagedPasteboard()
@@ -0,0 +1,183 @@
import Foundation
import TermindDomain
import TermindRemoteInfrastructure
import Testing
@Suite
struct SFTPConcurrencySchedulerTests {
@Test func smartResolvesTheSixtyFourMBBoundaryAndUnknownSizeToChannels() async throws {
#expect(SFTPConcurrencyStrategy.smartConnectionThreshold == 64_000_000)
let scheduler = SFTPConcurrencyScheduler(
preferences: SFTPPreferences(globalLimit: 3, perHostLimit: 3, strategy: .smart))
let host = UUID()
let source = SFTPWorkSource.workspace(UUID())
let unknown = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: source, byteCount: nil)
let boundary = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: source,
byteCount: SFTPConcurrencyStrategy.smartConnectionThreshold)
let large = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: source,
byteCount: SFTPConcurrencyStrategy.smartConnectionThreshold + 1)
#expect(unknown.schedulingMode == .channelsOnly)
#expect(boundary.schedulingMode == .channelsOnly)
#expect(large.schedulingMode == .connectionsPreferred)
await unknown.release()
await boundary.release()
await large.release()
}
@Test func sameHostRotatesFromWorkspaceToAgentRun() async throws {
let scheduler = SFTPConcurrencyScheduler(
preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 1, strategy: .channelsOnly))
let host = UUID()
let workspace = SFTPWorkSource.workspace(UUID())
let agent = SFTPWorkSource.agentRun(UUID())
let first = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: workspace, byteCount: 1)
let recorder = GrantRecorder()
let workspaceTask = queuedAcquire(
scheduler: scheduler, host: host, source: workspace, label: "workspace", recorder: recorder)
try await eventually { await scheduler.queuedCount == 1 }
let agentTask = queuedAcquire(
scheduler: scheduler, host: host, source: agent, label: "agent", recorder: recorder)
try await eventually { await scheduler.queuedCount == 2 }
await first.release()
try await eventually { await recorder.values.count == 1 }
let grants = await recorder.values
#expect(grants == ["agent"])
let winner = grants.first == "agent" ? agentTask : workspaceTask
let loser = grants.first == "agent" ? workspaceTask : agentTask
let winnerPermit = await winner.value
await winnerPermit?.release()
loser.cancel()
_ = await loser.value
}
@Test func hostsRotateBeforeSources() async throws {
let scheduler = SFTPConcurrencyScheduler(
preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 1, strategy: .channelsOnly))
let firstHost = UUID()
let secondHost = UUID()
let source = SFTPWorkSource.workspace(UUID())
let first = try await scheduler.acquire(
anchorHostID: firstHost, hostIDs: [firstHost], source: source, byteCount: 1)
let recorder = GrantRecorder()
let sameHostTask = queuedAcquire(
scheduler: scheduler, host: firstHost, source: source, label: "first", recorder: recorder)
try await eventually { await scheduler.queuedCount == 1 }
let otherHostTask = queuedAcquire(
scheduler: scheduler, host: secondHost, source: source, label: "second", recorder: recorder)
try await eventually { await scheduler.queuedCount == 2 }
await first.release()
try await eventually { await recorder.values.count == 1 }
let grants = await recorder.values
#expect(grants == ["second"])
let winner = grants.first == "second" ? otherHostTask : sameHostTask
let loser = grants.first == "second" ? sameHostTask : otherHostTask
let winnerPermit = await winner.value
await winnerPermit?.release()
loser.cancel()
_ = await loser.value
}
@Test func crossHostCopyUsesOneGlobalSlotAndOneSlotOnEachHost() async throws {
let scheduler = SFTPConcurrencyScheduler(
preferences: SFTPPreferences(globalLimit: 2, perHostLimit: 1, strategy: .channelsOnly))
let sourceHost = UUID()
let destinationHost = UUID()
let unrelatedHost = UUID()
let copy = try await scheduler.acquire(
anchorHostID: destinationHost,
hostIDs: [sourceHost, destinationHost],
source: .workspace(UUID()),
byteCount: 1)
let unrelated = try await scheduler.acquire(
anchorHostID: unrelatedHost,
hostIDs: [unrelatedHost],
source: .workspace(UUID()),
byteCount: 1)
#expect(await scheduler.activeCount == 2)
#expect(await scheduler.activeCount(for: sourceHost) == 1)
#expect(await scheduler.activeCount(for: destinationHost) == 1)
#expect(await scheduler.activeCount(for: unrelatedHost) == 1)
await copy.release()
await unrelated.release()
}
@Test func fiveSameHostSlotsSplitThreeTwoAcrossWorkspaces() async throws {
let scheduler = SFTPConcurrencyScheduler(
preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 5, strategy: .channelsOnly))
let host = UUID()
let blocker = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: .batch(UUID()), byteCount: 1)
let recorder = GrantRecorder()
let first = SFTPWorkSource.workspace(UUID())
let second = SFTPWorkSource.workspace(UUID())
var tasks: [Task<SFTPConcurrencyPermit?, Never>] = []
for _ in 0..<4 {
tasks.append(
queuedAcquire(
scheduler: scheduler, host: host, source: first,
label: "first", recorder: recorder))
tasks.append(
queuedAcquire(
scheduler: scheduler, host: host, source: second,
label: "second", recorder: recorder))
}
try await eventually { await scheduler.queuedCount == 8 }
await scheduler.updatePreferences(
SFTPPreferences(globalLimit: 5, perHostLimit: 5, strategy: .channelsOnly))
await blocker.release()
try await eventually { await recorder.values.count == 5 }
let grants = await recorder.values
#expect(grants.filter { $0 == "first" }.count == 3)
#expect(grants.filter { $0 == "second" }.count == 2)
for task in tasks { task.cancel() }
for task in tasks {
let permit = await task.value
await permit?.release()
}
}
private func queuedAcquire(
scheduler: SFTPConcurrencyScheduler,
host: UUID,
source: SFTPWorkSource,
label: String,
recorder: GrantRecorder
) -> Task<SFTPConcurrencyPermit?, Never> {
Task {
do {
let permit = try await scheduler.acquire(
anchorHostID: host, hostIDs: [host], source: source, byteCount: 1)
await recorder.append(label)
return permit
} catch {
return nil
}
}
}
private func eventually(_ condition: () async -> Bool) async throws {
for _ in 0..<500 {
if await condition() { return }
try await Task.sleep(for: .milliseconds(10))
}
try #require(await condition())
}
}
private actor GrantRecorder {
private(set) var values: [String] = []
func append(_ value: String) { values.append(value) }
}
+26 -3
View File
@@ -124,7 +124,7 @@ struct SFTPCopyToolTests {
@Test
func sameHostUsesIndependentConnectionsAndRejectsSelfCopies() async throws {
let fixture = CopyFixture(sameHost: true)
let fixture = CopyFixture(sameHost: true, strategy: .connectionsPreferred)
await fixture.source.directory("/target")
await fixture.source.put("/source/a.bin", bytes: Data([1]))
#expect(try await !fixture.copy().isError)
@@ -139,6 +139,17 @@ struct SFTPCopyToolTests {
#expect(await fixture.source.nodes["/source/tree/nested/tree"] == nil)
}
@Test
func smartUnknownSameHostCopyUsesOneChannel() async throws {
let fixture = CopyFixture(sameHost: true)
await fixture.source.directory("/target")
await fixture.source.put("/source/a.bin", bytes: Data([1]))
#expect(try await !fixture.copy().isError)
#expect(await fixture.backend.opens == 1)
#expect(await fixture.backend.siblingChannels == 0)
}
@Test
func detectsChangedSourceAndReportsPartialCompletionWithoutRetry() async throws {
let fixture = CopyFixture()
@@ -333,7 +344,10 @@ private final class CopyFixture {
let authentication: AgentSSHAuthenticationRegistry
var tool: SFTPCopyTool { .init(hosts: resolver, authentication: authentication, connections: pool) }
init(sameHost: Bool = false) {
init(
sameHost: Bool = false,
strategy: SFTPConcurrencyStrategy = .smart
) {
destinationID = sameHost ? sourceID : UUID()
let sourceID = sourceID
let destinationID = destinationID
@@ -362,7 +376,10 @@ private final class CopyFixture {
pool = AgentConnectionPool(
configuration: { configuration($0) }, routeFingerprint: { _ in "test" },
openExec: { _ in throw AgentToolError.invalidHost },
openSFTP: { try await backend.open($0.configuration.hostID) }, sweepInterval: nil
openSFTP: { try await backend.open($0.configuration.hostID) },
sftpConcurrencyScheduler: SFTPConcurrencyScheduler(
preferences: SFTPPreferences(strategy: strategy)),
sweepInterval: nil
)
}
@@ -457,12 +474,14 @@ private actor CopyBackend {
var filesystems: [UUID: CopyFileSystem]
var opens = 0
var disconnects = 0
var siblingChannels = 0
init(filesystems: [UUID: CopyFileSystem]) { self.filesystems = filesystems }
func open(_ id: UUID) throws -> CopyClient {
opens += 1
return CopyClient(filesystem: try #require(filesystems[id]), backend: self)
}
func disconnected() { disconnects += 1 }
func openedSibling() { siblingChannels += 1 }
}
private actor CopyClient: SFTPClientProtocol {
@@ -473,6 +492,10 @@ private actor CopyClient: SFTPClientProtocol {
self.backend = backend
}
func connect(using configuration: SSHConnectionConfiguration, progress: SSHRouteProgressHandler?) async throws {}
func openSiblingChannel() async throws -> any SFTPClientProtocol {
await backend.openedSibling()
return CopyClient(filesystem: filesystem, backend: backend)
}
func disconnect() async { await backend.disconnected() }
func homeDirectory() async throws -> String { "/" }
func listDirectory(at path: String) async throws -> [RemoteFile] { await filesystem.children(path) }
+7 -3
View File
@@ -12,15 +12,19 @@ struct SFTPSettingsIntegrationTests {
let settings = SFTPSettingsModel()
let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings)
defer { app.shutdownForTermination() }
#expect(app.sftpTransfers.globalLimit == 15)
#expect(app.sftpTransfers.perHostLimit == 5)
#expect(app.sftpTransfers.globalLimit == 30)
#expect(app.sftpTransfers.perHostLimit == 10)
#expect(app.sftpTransfers.strategy == .smart)
#expect(settings.set(100, for: \.globalLimit))
#expect(settings.set(50, for: \.perHostLimit))
#expect(settings.setStrategy(.connectionsPreferred))
#expect(app.sftpTransfers.globalLimit == 100)
#expect(app.sftpTransfers.perHostLimit == 50)
settings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2))
#expect(app.sftpTransfers.strategy == .connectionsPreferred)
settings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2, strategy: .channelsOnly))
#expect(app.sftpTransfers.globalLimit == 3)
#expect(app.sftpTransfers.perHostLimit == 2)
#expect(app.sftpTransfers.strategy == .channelsOnly)
}
@Test(.requiresVisibleWindow, arguments: ["en", "zh-Hans", "zh-Hant"])
+22 -5
View File
@@ -14,7 +14,8 @@ struct SFTPTransferSessionTests {
let configuration = SSHConnectionConfiguration(
hostID: UUID(), host: "fixture", port: 22, username: "user", credentials: [], expectedHostKeys: [])
let coordinator = SFTPTransferCoordinator(
store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) })
store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) },
strategy: .connectionsPreferred)
let browser = SFTPSessionModel(
hostID: configuration.hostID, initialPath: "/", client: primary, configuration: configuration)
let workspace = Workspace(pane: Pane(.sftpSession(browser)))
@@ -88,6 +89,21 @@ struct SFTPTransferSessionTests {
#expect(await harness.opens == 1)
}
@Test func channelsOnlyUsesSiblingChannelsWithoutOpeningConnections() async throws {
let harness = TransferSessionHarness(behaviors: [])
let primary = TransferSessionClient(harness: harness, primary: true)
let (session, coordinator) = fixture(
primary: primary, harness: harness, limit: 3, strategy: .channelsOnly)
enqueue(3, session: session, coordinator: coordinator)
try await eventually { await harness.started == 3 }
#expect(await harness.opens == 0)
#expect(await harness.siblingChannels == 3)
#expect(await harness.maximumActive == 3)
await harness.releaseAll()
try await eventually { coordinator.activeCount == 0 }
await session.releaseOwner()
}
@Test func emptyCredentialsCanOpenParallelConnectionsWithinTheLimits() async throws {
let harness = TransferSessionHarness(behaviors: [])
let primary = TransferSessionClient(harness: harness, primary: true)
@@ -180,7 +196,7 @@ struct SFTPTransferSessionTests {
hostID: UUID(), host: "terminal", port: 22, username: "user", credentials: [], expectedHostKeys: [])
let coordinator = SFTPTransferCoordinator(
store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) },
globalLimit: 15, perHostLimit: 5)
globalLimit: 15, perHostLimit: 5, strategy: .connectionsPreferred)
let upload = SFTPSessionModel(
hostID: configuration.hostID, initialPath: "/destination", client: primary,
configuration: configuration, wasOpenedSilently: true)
@@ -215,7 +231,7 @@ struct SFTPTransferSessionTests {
let harness = TransferSessionHarness(behaviors: [])
let coordinator = SFTPTransferCoordinator(
store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) },
globalLimit: 15, perHostLimit: 5)
globalLimit: 15, perHostLimit: 5, strategy: .connectionsPreferred)
let configuration = SSHConnectionConfiguration(
hostID: UUID(), host: "same-host", port: 22, username: "user", credentials: [], expectedHostKeys: [])
let sessions = (0..<2).map { _ in
@@ -237,7 +253,8 @@ struct SFTPTransferSessionTests {
private func fixture(
primary: TransferSessionClient, harness: TransferSessionHarness, limit: Int = 4,
wasOpenedSilently: Bool = false
wasOpenedSilently: Bool = false,
strategy: SFTPConcurrencyStrategy = .connectionsPreferred
) -> (SFTPTransferSession, SFTPTransferCoordinator) {
let session = SFTPTransferSession(
configuration: SSHConnectionConfiguration(
@@ -246,7 +263,7 @@ struct SFTPTransferSessionTests {
client: primary, wasOpenedSilently: wasOpenedSilently)
let coordinator = SFTPTransferCoordinator(
store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) },
globalLimit: limit, perHostLimit: limit)
globalLimit: limit, perHostLimit: limit, strategy: strategy)
return (session, coordinator)
}