Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
02ed751
refactor: isolate surface catalog query ownership
austinywang Sep 10, 2026
39920fb
test: reproduce new Cloud machine catalog discovery race
austinywang Sep 10, 2026
3636250
fix: discover new Cloud machines before reading their catalog
austinywang Sep 10, 2026
b4da292
fix: race cloud private addresses before choosing a connection path
austinywang Sep 10, 2026
6d11b79
Merge remote-tracking branch 'origin/fix-cloud-snapshot-connect-readi…
austinywang Sep 10, 2026
4efb148
fix: serve the cloud desktop over both private address families
austinywang Sep 10, 2026
3560644
fix: reuse the working family for successive cloud port connections
austinywang Sep 10, 2026
cb672b4
fix: record verified dual-stack desktop snapshot ladder
austinywang Sep 10, 2026
3f43aff
refactor: inject Cloud discovery and provider refresh operations
austinywang Sep 10, 2026
c4f3152
test: cover Cloud discovery stalls and abandoned hub dials
austinywang Sep 10, 2026
1446992
test: encode the hub fixture IPv6 address explicitly
austinywang Sep 10, 2026
a977941
fix: bound Cloud discovery and cancel abandoned tunnel dials
austinywang Sep 10, 2026
028c4be
test: reproduce stale primary route with a sole IPv6 candidate
austinywang Sep 10, 2026
5a3b354
Merge remote-tracking branch 'origin/fix-cloud-snapshot-connect-readi…
austinywang Sep 10, 2026
550d841
test: cover discovery retirement and browser address reuse
austinywang Sep 10, 2026
d6a4127
test: split browser address reuse coverage into its own suite
austinywang Sep 10, 2026
406748a
fix: retire Cloud discovery waiters and preserve sibling refreshes
austinywang Sep 10, 2026
c9461e9
Merge remote-tracking branch 'origin/fix-cloud-snapshot-connect-readi…
austinywang Sep 10, 2026
39b6c04
fix: use the sole current private address instead of a stale route
austinywang Sep 10, 2026
2c558f0
Merge origin/main and reconcile Cloud test project entries
austinywang Sep 10, 2026
b93855d
Merge branch 'main' of https://github.com/manaflow-ai/cmux into issue…
austinywang Sep 10, 2026
d3e0d7a
test: cover partial Cloud routes and post-sign-out discovery
austinywang Sep 10, 2026
6c9200d
fix: preserve Cloud fallback routes and serialize account discovery
austinywang Sep 10, 2026
bb8a40a
test: isolate Cloud registry notifications across parallel suites
austinywang Sep 10, 2026
68d7c50
test: reproduce publishing a Cloud VM before network announcement
austinywang Sep 10, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Sources/Auth/MacAuthComposition.swift
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ struct MacAuthComposition {
browserAppSessionSignInRelay.sessionWillTransition()
},
onSignedIn: {
await CmuxTuiSurfaceProviderRegistry.shared.resumeAfterSignIn()
await browserAppSessionSignInRelay.signedIn()
}
)
Expand Down
53 changes: 53 additions & 0 deletions Sources/Cloud/CloudMachineLinkManager+PrivateRoute.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
import Foundation
import Network

extension CloudMachineLinkManager {
/// Selects a working family using the same SOCKS connector as port forwards.
/// The probe sends no daemon request and closes its stream before the real
/// encrypted link starts. Failed probes propagate to the caller's retry policy,
/// so the next attempt considers every current family again.
func resolvedPrivateRoute(
machineID: String,
through hub: CloudWireGuardHub.Ready,
fallbackRoute: String? = nil,
addresses freshAddresses: [String] = []
) async throws -> String {
try Task.checkCancellation()
let freshAddresses = freshAddresses.map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
.filter { !$0.isEmpty }
let freshFamilies = Set(freshAddresses.map { $0.contains(":") })
// Attach can omit one family. Preserve that family from discovery, but
// never retain an old address in a family that attach has replaced.
let storedAddresses = privateAddresses(for: machineID).filter {
!freshFamilies.contains($0.contains(":"))
}
var seen = Set<String>()
let candidates = (freshAddresses + storedAddresses).filter { seen.insert($0).inserted }
let addresses = candidates.filter {
CloudWireGuardHub.routesHost($0, enrolledRoutes: hub.routes)
}
guard let primary = addresses.first else {
guard candidates.isEmpty,
let route = fallbackRoute ?? privateRoute(for: machineID),
let host = IPNetworkPrefix.routeHost(route),
CloudWireGuardHub.routesHost(host, enrolledRoutes: hub.routes) else {
throw ManagerError.privateRouteRequired(machineID)
}
return route
}
// Fresh discovery or the enrolled routes can leave a single candidate.
// Use it directly rather than returning an older address-family route.
guard addresses.count > 1 else {
let host = primary.contains(":") ? "[\(primary)]" : primary
return "ws://\(host):1337/v1/link"
Comment thread
cursor[bot] marked this conversation as resolved.
}
let connected = try await CloudHubConnector().connect(
endpoint: .unix(path: hub.socketPath),
target: CloudPortForwardTarget(host: primary, port: 1337, fallbackHosts: Array(addresses.dropFirst())),
queue: DispatchQueue.global(qos: .userInitiated)
)
connected.connection.cancel()
let host = connected.host.contains(":") ? "[\(connected.host)]" : connected.host
return "ws://\(host):1337/v1/link"
}
}
46 changes: 27 additions & 19 deletions Sources/Cloud/CloudMachineLinkManager.swift
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ actor CloudMachineLinkManager {
/// Private routes come from the signed-in machine list. An enrolled client
/// reconnects with this local fact and does not call the attach endpoint.
private var privateRoutes: [String: String] = [:]
private var privateAddressCandidates: [String: [String]] = [:]
private var links: [String: CloudMachineLink] = [:]
private var connecting: [String: Task<CloudMachineLink.Connected, Error>] = [:]
private var lastFailure: [String: (at: Date, error: String)] = [:]
Expand Down Expand Up @@ -97,14 +98,26 @@ actor CloudMachineLinkManager {
var hasClient: Bool { clientURL != nil }

func setPrivateAddress(_ address: String?, for machineID: String) {
guard let address = address?.trimmingCharacters(in: .whitespacesAndNewlines), !address.isEmpty else {
setPrivateAddresses(address.map { [$0] } ?? [], for: machineID)
}

func setPrivateAddresses(_ addresses: [String], for machineID: String) {
var seen = Set<String>()
let addresses = addresses.map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
.filter { !$0.isEmpty && seen.insert($0).inserted }
privateAddressCandidates[machineID] = addresses
guard let address = addresses.first else {
privateRoutes[machineID] = nil
return
}
let host = address.contains(":") ? "[\(address)]" : address
privateRoutes[machineID] = "ws://\(host):1337/v1/link"
}

func privateAddresses(for machineID: String) -> [String] {
privateAddressCandidates[machineID] ?? []
}

func privateRoute(for machineID: String) -> String? {
privateRoutes[machineID]
}
Expand All @@ -121,7 +134,7 @@ actor CloudMachineLinkManager {
throw ManagerError.retryLater(failure.error)
}
guard let clientURL else { throw ManagerError.clientMissing }
guard let privateRoute = privateRoutes[machineID] else {
guard privateRoutes[machineID] != nil else {
throw ManagerError.privateRouteRequired(machineID)
}
#if DEBUG
Expand Down Expand Up @@ -164,26 +177,22 @@ actor CloudMachineLinkManager {
guard capabilities.contains(CloudTuiCommandLine.wireGuardHubCapability) else {
throw ManagerError.wireGuardHubUnsupported
}
guard Self.usesWireGuardHub(route: privateRoute, clientCapabilities: capabilities, enrolledRoutes: []) else {
throw ManagerError.privateRouteRequired(privateRoute)
}
guard let hub else { throw ManagerError.wireGuardHubMissing }
let claim = try await hub.acquire()
guard Self.usesWireGuardHub(
route: privateRoute,
clientCapabilities: capabilities,
enrolledRoutes: claim.ready.routes
) else {
await hub.release(claim.lease)
throw ManagerError.privateRouteRequired(privateRoute)
}
let releaseLease: @Sendable () async -> Void = { await hub.release(claim.lease) }
let reachableRoute: String
do {
reachableRoute = try await resolvedPrivateRoute(machineID: machineID, through: claim.ready)
Comment thread
cursor[bot] marked this conversation as resolved.
} catch {
await releaseLease()
throw error
}
#if DEBUG
cmuxDebugLog("cloud.link.wireguardHub machine=\(machineID) socket=\(claim.ready.socketPath)")
#endif
let connect = Task {
try await link.connect(
route: privateRoute,
route: reachableRoute,
session: session,
carrier: carrier,
timeout: connectTimeout,
Expand Down Expand Up @@ -275,12 +284,11 @@ actor CloudMachineLinkManager {
lastFailure.removeAll()
}

/// Drops links for machines that no longer exist.
func retain(machineIDs: Set<String>) async {
for id in links.keys where !machineIDs.contains(id) {
await disconnect(machineID: id)
}
/// Drops stale routing facts immediately. The registry owns and awaits
/// each removed machine's asynchronous link/forward teardown separately.
func retainAddresses(machineIDs: Set<String>) {
privateRoutes = privateRoutes.filter { machineIDs.contains($0.key) }
privateAddressCandidates = privateAddressCandidates.filter { machineIDs.contains($0.key) }
}

/// Re-sends this Mac's theme to every connected machine (a Ghostty config reload
Expand Down
7 changes: 7 additions & 0 deletions Sources/Cloud/PortForward/CloudHubConnection.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
import Network

/// The successful private-address dial and the hub stream carrying its bytes.
struct CloudHubConnection: Sendable {
let connection: NWConnection
let host: String
}
93 changes: 93 additions & 0 deletions Sources/Cloud/PortForward/CloudHubConnector.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
import Foundation
import Network

/// Connects to the first working private address through one claimed hub.
/// A family can blackhole independently of the other after a VM joins its VPC.
/// Race actual SOCKS CONNECT handshakes, retaining the winning stream and closing
/// every loser before returning, so terminal and browser callers share the policy.
struct CloudHubConnector: Sendable {
var timeout: Duration = .seconds(15)
/// A cancellable head start for the preferred family, driven by the injected clock.
var fallbackDelay: Duration = .milliseconds(250)
var clock: any Clock<Duration> = ContinuousClock()

#if compiler(>=6.2)
@concurrent
#else
@Sendable
#endif
func connect(
endpoint: NWEndpoint,
target: CloudPortForwardTarget,
queue: DispatchQueue
) async throws -> CloudHubConnection {
let candidates = target.hosts.map { host in
CloudHubConnection(connection: NWConnection(to: endpoint, using: .tcp), host: host)
}
return try await withTaskCancellationHandler {
try await withThrowingTaskGroup(of: Result<CloudHubConnection, any Error>.self) { group in
for (index, candidate) in candidates.enumerated() {
group.addTask {
do {
if index > 0 { try await clock.sleep(for: fallbackDelay) }
try Task.checkCancellation()
try await handshake(candidate.connection, host: candidate.host, port: target.port, queue: queue)
try Task.checkCancellation()
return .success(candidate)
} catch {
candidate.connection.cancel()
return .failure(error)
}
}
}
defer { group.cancelAll() }
var lastError: any Error = CancellationError()
while let result = try await group.next() {
switch result {
case .success(let connected):
for other in candidates where other.connection !== connected.connection {
other.connection.cancel()
}
if Task.isCancelled {
connected.connection.cancel()
throw CancellationError()
}
return connected
case .failure(let error):
lastError = error
}
}
throw lastError
}
} onCancel: {
for candidate in candidates { candidate.connection.cancel() }
}
}

#if compiler(>=6.2)
@concurrent
#else
@Sendable
#endif
private func handshake(_ connection: NWConnection, host: String, port: Int, queue: DispatchQueue) async throws {
try await withTaskCancellationHandler {
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
try await connection.startAndWaitUntilReady(queue: queue)
try await CloudPortForwardRelay.connect(connection, to: CloudPortForwardTarget(host: host, port: port))
}
group.addTask {
// A real handshake deadline; completion cancels this child
// and expiry cancels the socket to unblock Network callbacks.
try await clock.sleep(for: timeout)
connection.cancel()
throw CloudPortForwardRelay.RelayError.handshakeTimedOut(timeout)
}
defer { group.cancelAll() }
try await group.next()
}
} onCancel: {
connection.cancel()
}
}
}
21 changes: 19 additions & 2 deletions Sources/Cloud/PortForward/CloudLoopbackPortForward.swift
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,9 @@ actor CloudLoopbackPortForward {
}

private(set) var target: CloudPortForwardTarget
/// Prefer the last successful family for later HTTP/WebSocket connections;
/// a noVNC asset burst must not redial a blackholed family for every file.
private var preferredHost: String?
/// The bound loopback port; 0 until ``start()`` returns.
private(set) var localPort: UInt16 = 0
private(set) var isListening = false
Expand Down Expand Up @@ -123,6 +126,7 @@ actor CloudLoopbackPortForward {
/// changes.
func retarget(_ newTarget: CloudPortForwardTarget) {
target = newTarget
if let preferredHost, !newTarget.hosts.contains(preferredHost) { self.preferredHost = nil }
}

/// Stops accepting and ends every connection. Returns once the listener
Expand Down Expand Up @@ -154,7 +158,12 @@ actor CloudLoopbackPortForward {
let id = UUID()
connections[id] = connection
acceptedConnectionCount += 1
let target = self.target
let target: CloudPortForwardTarget
if let preferredHost {
target = CloudPortForwardTarget(host: preferredHost, port: self.target.port, fallbackHosts: self.target.hosts)
} else {
target = self.target
}
let relay = self.relay
let queue = self.queue
Task { [weak self] in
Expand All @@ -165,11 +174,19 @@ actor CloudLoopbackPortForward {
await self?.connectionDidEnd(id)
return
}
await relay.carry(connection, to: target, queue: queue)
await relay.carry(connection, to: target, queue: queue) { [weak self] host in
await self?.rememberSuccessfulHost(host, for: target)
}
await self?.connectionDidEnd(id)
}
}

private func rememberSuccessfulHost(_ host: String, for attemptedTarget: CloudPortForwardTarget) {
guard !stopped, target.port == attemptedTarget.port,
Set(target.hosts) == Set(attemptedTarget.hosts) else { return }
preferredHost = host
}

private func connectionDidEnd(_ id: UUID) {
connections[id] = nil
}
Expand Down
36 changes: 11 additions & 25 deletions Sources/Cloud/PortForward/CloudPortForwardRelay.swift
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,12 @@ struct CloudPortForwardRelay: Sendable {

/// Returns when the connection has ended. A failure before the relay begins
/// closes `client`, so a browser sees a connection error rather than a hang.
func carry(_ client: NWConnection, to target: CloudPortForwardTarget, queue: DispatchQueue) async {
func carry(
_ client: NWConnection,
to target: CloudPortForwardTarget,
queue: DispatchQueue,
onConnected: @Sendable (String) async -> Void = { _ in }
) async {
let claim: CloudHubSocketClaim
do {
claim = try await dialer.claimHubSocket()
Expand All @@ -38,12 +43,14 @@ struct CloudPortForwardRelay: Sendable {
client.cancel()
return
}
let upstream = NWConnection(to: claim.endpoint, using: .tcp)
let upstream: NWConnection
do {
try await handshake(upstream, to: target, queue: queue)
let connected = try await CloudHubConnector(timeout: handshakeTimeout, clock: clock)
.connect(endpoint: claim.endpoint, target: target, queue: queue)
upstream = connected.connection
await onConnected(connected.host)
} catch {
logger.error("SOCKS5 CONNECT to \(target.host, privacy: .private):\(target.port, privacy: .public) failed: \(CloudMachineLink.errorText(error), privacy: .public)")
upstream.cancel()
client.cancel()
await claim.release()
return
Expand All @@ -52,27 +59,6 @@ struct CloudPortForwardRelay: Sendable {
await claim.release()
}

/// The hub connection and SOCKS5 exchange under ``handshakeTimeout``. The
/// deadline cancels the connection, which is what unblocks a stalled
/// receive; the loser of the race is discarded.
private func handshake(_ upstream: NWConnection, to target: CloudPortForwardTarget, queue: DispatchQueue) async throws {
let budget = handshakeTimeout
let clock = self.clock
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
try await upstream.startAndWaitUntilReady(queue: queue)
try await Self.connect(upstream, to: target)
}
group.addTask {
try await clock.sleep(for: budget)
upstream.cancel()
throw RelayError.handshakeTimedOut(budget)
}
defer { group.cancelAll() }
try await group.next()
}
}

/// The SOCKS5 handshake on a ready hub connection; on return the stream
/// carries the tunneled connection.
static func connect(_ upstream: NWConnection, to target: CloudPortForwardTarget) async throws {
Expand Down
8 changes: 8 additions & 0 deletions Sources/Cloud/PortForward/CloudPortForwardTarget.swift
Original file line number Diff line number Diff line change
Expand Up @@ -5,4 +5,12 @@ import Foundation
struct CloudPortForwardTarget: Sendable, Hashable {
let host: String
let port: Int
/// Other addresses advertised for this same machine. They are raced through
/// the same hub; the private network remains the only route.
var fallbackHosts: [String] = []

var hosts: [String] {
var seen = Set<String>()
return ([host] + fallbackHosts).filter { !$0.isEmpty && seen.insert($0).inserted }
}
}
Loading
Loading