diff --git a/Sources/Surfaces/CloudProviderRefreshCoordinator.swift b/Sources/Surfaces/CloudProviderRefreshCoordinator.swift new file mode 100644 index 000000000000..36140fe7eb41 --- /dev/null +++ b/Sources/Surfaces/CloudProviderRefreshCoordinator.swift @@ -0,0 +1,55 @@ +import Foundation + +/// Serializes graph publication for one provider. A forced reader waits for a +/// pass started after its request; ordinary readers share the active pass. +@MainActor +final class CloudProviderRefreshCoordinator { + private struct Entry { + let request: UInt64 + let forced: Bool + let task: Task + } + + private var inFlight: Entry? + private var latestRequest: UInt64 = 0 + private var invalidation: UInt64 = 0 + private var lifetime: UInt64 = 0 + + func refresh(force: Bool, operation: @escaping @MainActor (Bool) async -> Bool) async -> Bool { + latestRequest &+= 1 + let request = latestRequest + let epoch = lifetime + while !Task.isCancelled, epoch == lifetime { + if let entry = inFlight { + let result = await entry.task.value + if inFlight?.task == entry.task { inFlight = nil } + guard !Task.isCancelled, epoch == lifetime else { return false } + if !force || (entry.forced && entry.request >= request) { return result } + continue + } + let task = Task { @MainActor [weak self] in + while let self, !Task.isCancelled, epoch == self.lifetime { + let revision = self.invalidation + let result = await operation(force) + guard !Task.isCancelled, epoch == self.lifetime else { return false } + // Metadata superseded this pass. Readers stay attached to + // the owner until a pass over the current metadata finishes. + if revision == self.invalidation { return result } + } + return false + } + // Covers all forced readers already waiting, so a burst shares + // one trailing pass instead of issuing one snapshot per waiter. + inFlight = Entry(request: latestRequest, forced: force, task: task) + } + return false + } + + func invalidate() { invalidation &+= 1 } + + func cancel() { + lifetime &+= 1 + inFlight?.task.cancel() + inFlight = nil + } +} diff --git a/Sources/Surfaces/CloudVMState+SnapshotComparison.swift b/Sources/Surfaces/CloudVMState+SnapshotComparison.swift new file mode 100644 index 000000000000..27edda5673f6 --- /dev/null +++ b/Sources/Surfaces/CloudVMState+SnapshotComparison.swift @@ -0,0 +1,62 @@ +import Foundation + +extension CloudVMState { + static func == (lhs: CloudVMState, rhs: CloudVMState) -> Bool { + lhs.hasSameModeledContent(as: rhs) && lhs.document == rhs.document + } + + /// Clients and live terminal titles and dimensions are observations, not + /// revisioned resources (client.list and public_terminal_snapshot). Keep + /// them in exports without treating inspection or resize as a conflict. + func hasSameRevisionedContent(as other: CloudVMState) -> Bool { + hasSameModeledContent(as: other, includingLiveTerminalMetadata: false) + && document.values.filter { $0.key != "clients" } == other.document.values.filter { $0.key != "clients" } + && document.collections.filter { $0.key != "clients" && $0.key != "terminals" } + == other.document.collections.filter { $0.key != "clients" && $0.key != "terminals" } + && hasSameTerminalDocument(as: other) + } + + private var revisionedTerminals: [CloudVMTerminalState] { + terminals.map { + var terminal = $0 + terminal.title = "" + terminal.cols = nil + terminal.rows = nil + return terminal + } + } + + private func hasSameModeledContent(as other: CloudVMState, includingLiveTerminalMetadata: Bool = true) -> Bool { + let left = includingLiveTerminalMetadata ? terminals : revisionedTerminals + let right = includingLiveTerminalMetadata ? other.terminals : other.revisionedTerminals + return machine == other.machine + && cursor == other.cursor + && workspaces == other.workspaces + && screens == other.screens + && panes == other.panes + && tabs == other.tabs + && left == right + && browsers == other.browsers + && agents == other.agents + } + + /// Identity, launch fields, and unknown fields remain strict. Only the PTY + /// title and dimensions are live; unchanged rows use their byte cache. + private func hasSameTerminalDocument(as other: CloudVMState) -> Bool { + guard let left = document.collections["terminals"] else { + return other.document.collections["terminals"] == nil + } + guard let right = other.document.collections["terminals"], left.order == right.order else { return false } + for id in left.order { + guard let a = left.rows[id], let b = right.rows[id] else { return false } + if a == b { continue } + guard var lhs = try? JSONSerialization.jsonObject(with: a) as? [String: Any], + var rhs = try? JSONSerialization.jsonObject(with: b) as? [String: Any] else { return false } + for key in ["title", "cols", "rows"] { lhs[key] = nil; rhs[key] = nil } + guard let lhsData = try? JSONSerialization.data(withJSONObject: lhs, options: [.sortedKeys]), + let rhsData = try? JSONSerialization.data(withJSONObject: rhs, options: [.sortedKeys]), + lhsData == rhsData else { return false } + } + return true + } +} diff --git a/Sources/Surfaces/CmuxTuiSurfaceProvider+PortForward.swift b/Sources/Surfaces/CmuxTuiSurfaceProvider+PortForward.swift index 84d8d4bddd00..f659d92b0974 100644 --- a/Sources/Surfaces/CmuxTuiSurfaceProvider+PortForward.swift +++ b/Sources/Surfaces/CmuxTuiSurfaceProvider+PortForward.swift @@ -12,7 +12,8 @@ extension CmuxTuiSurfaceProvider { func materializeBrowserPane( _ resource: SurfaceResource, at destination: SurfaceDestination, - focus: Bool + focus: Bool, + reusing existingPane: (workspaceID: UUID, panelID: UUID)? = nil ) async throws -> (workspaceID: UUID, panelID: UUID) { let generation = currentLifecycleGeneration try Task.checkCancellation() @@ -33,7 +34,7 @@ extension CmuxTuiSurfaceProvider { try Task.checkCancellation() guard isCurrentLifecycleGeneration(generation), isRegisteredInCatalog() else { throw CancellationError() } let label = Self.paneLabel(machineID: machineID, port: target.port, desktop: desktop) - let pane = try Self.makeConnectingPane(label: label, at: destination, focus: focus) + let pane = try Self.makeConnectingPane(label: label, at: destination, focus: focus, reusing: existingPane) let machineWasAwake = isAwake // A provider that is stopped or replaced while this runs must not // touch the pane its successor now owns. @@ -66,7 +67,7 @@ extension CmuxTuiSurfaceProvider { case .controlPlanePreview(let port): guard isRegisteredInCatalog() else { throw CancellationError() } let label = Self.paneLabel(machineID: machineID, port: port, desktop: desktop) - let pane = try Self.makeConnectingPane(label: label, at: destination, focus: focus) + let pane = try Self.makeConnectingPane(label: label, at: destination, focus: focus, reusing: existingPane) browserPaneTasks[pane.panelID] = Task { @MainActor [weak self] in guard let self else { return } defer { self.browserPaneTasks[pane.panelID] = nil } @@ -86,6 +87,44 @@ extension CmuxTuiSurfaceProvider { } } + /// Restored browser tabs retain their identity, but their saved loopback + /// ports belong to the previous process. Reuse the normal route preparation + /// path to create a new forward and navigate the existing tab in place. + func reprojectRestoredBrowserPanes(generation: UInt64) { + for resource in catalog.snapshot.resources(on: machine) where resource.kind != .terminal { + for projection in catalog.projections(of: resource.id) + where !materializedPanels.contains(projection.panelID) { + guard SurfacePaneFactory.browserPanel(panelID: projection.panelID, in: projection.workspaceID) != nil, + let paneID = SurfacePaneFactory.paneID(ofPanel: projection.panelID, in: projection.workspaceID) else { continue } + materializedPanels.insert(projection.panelID) + let pane = (workspaceID: projection.workspaceID, panelID: projection.panelID) + // The old process no longer owns this URL. Retire it before + // any route setup can suspend, then reuse the normal preparer. + SurfacePaneFactory.navigate(panelID: pane.panelID, in: pane.workspaceID, to: SurfacePaneFactory.blankURL) + SurfacePaneFactory.showPlaceholder(SurfaceBrowserPlaceholder.connecting(resource.title), panelID: pane.panelID, in: pane.workspaceID) + // This task owns forward creation; materializeBrowserPane hands + // the same slot to its navigation task after the forward binds. + browserPaneTasks[pane.panelID] = Task { @MainActor [weak self] in + guard let self else { return } + do { + try Task.checkCancellation() + guard self.isCurrentLifecycleGeneration(generation), self.isRegisteredInCatalog() else { return } + _ = try await self.materializeBrowserPane( + resource, + at: .tab(workspaceID: pane.workspaceID, paneID: paneID, index: nil), + focus: false, + reusing: pane + ) + } catch { + self.browserPaneTasks[pane.panelID] = nil + guard !Task.isCancelled, self.isCurrentLifecycleGeneration(generation) else { return } + Self.showFailure(label: resource.title, error: error, pane: pane) + } + } + } + } + } + /// The link `port` opens as, shared by the pane, Copy Link, and /// `vm.port_open`: the loopback forward when the machine has a private /// address, else the control plane's tokened preview URL. Throws when the @@ -138,9 +177,10 @@ extension CmuxTuiSurfaceProvider { private static func makeConnectingPane( label: String, at destination: SurfaceDestination, - focus: Bool + focus: Bool, + reusing existingPane: (workspaceID: UUID, panelID: UUID)? = nil ) throws -> (workspaceID: UUID, panelID: UUID) { - let pane = try SurfacePaneFactory.makeBrowserPane(url: SurfacePaneFactory.blankURL, at: destination, focus: focus) + let pane = try existingPane ?? SurfacePaneFactory.makeBrowserPane(url: SurfacePaneFactory.blankURL, at: destination, focus: focus) SurfacePaneFactory.showPlaceholder(SurfaceBrowserPlaceholder.connecting(label), panelID: pane.panelID, in: pane.workspaceID) return pane } diff --git a/Sources/Surfaces/CmuxTuiSurfaceProvider+Refresh.swift b/Sources/Surfaces/CmuxTuiSurfaceProvider+Refresh.swift new file mode 100644 index 000000000000..dab865c5fbe6 --- /dev/null +++ b/Sources/Surfaces/CmuxTuiSurfaceProvider+Refresh.swift @@ -0,0 +1,23 @@ +import Foundation + +extension CmuxTuiSurfaceProvider { + func refresh() async { + await refreshCurrentGraph(force: false) + } + + // Matches the protocol's Void return type so existential catalog reads + // preserve force instead of falling through to its legacy default. + func refresh(force: Bool) async { + await refreshCurrentGraph(force: force) + } + + /// Re-syncs the graph and reports whether the result is authoritative enough + /// for mutations. Concurrent reads share the provider's refresh owner. + @discardableResult + func refreshCurrentGraph(force: Bool) async -> Bool { + await refreshCoordinator.refresh(force: force) { [weak self] force in + guard let self else { return false } + return await self.performRefresh(force: force) + } + } +} diff --git a/Sources/Surfaces/CmuxTuiSurfaceProviders.swift b/Sources/Surfaces/CmuxTuiSurfaceProviders.swift index 40d0fba62225..b1553018bafa 100644 --- a/Sources/Surfaces/CmuxTuiSurfaceProviders.swift +++ b/Sources/Surfaces/CmuxTuiSurfaceProviders.swift @@ -25,6 +25,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { private var lifecycleGeneration: UInt64 = 0 /// Invalidates an older refresh before it can publish over a newer one. private var refreshGeneration: UInt64 = 0 + let refreshCoordinator = CloudProviderRefreshCoordinator() /// The only installed daemon graph for this machine. The catalog receives the /// same immutable value with its derived rows in one transaction. private(set) var cloudState: CloudVMState? @@ -122,6 +123,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { func update(summary: VMSummary) { guard isRegisteredInCatalog() else { return } refreshGeneration &+= 1 + refreshCoordinator.invalidate() self.summary = summary if !supportsPortPreviews { portsCache = nil @@ -144,6 +146,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { } func stop() { lifecycleGeneration &+= 1 + refreshCoordinator.cancel() for task in browserPaneTasks.values { task.cancel() } browserPaneTasks.removeAll() refreshGeneration &+= 1 @@ -190,14 +193,8 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { && refreshGeneration == refresh && isRegisteredInCatalog() } - // MARK: - SurfaceProvider - func refresh() async { - await refresh(force: false) - } - /// Re-syncs from the machine. A sleeping machine is never woken to be listed: it keeps - /// its screen (opening it wakes the machine) and nothing else. - @discardableResult - func refresh(force: Bool) async -> Bool { + /// One refresh pass. Sleeping machines retain their graph without being woken. + func performRefresh(force: Bool) async -> Bool { let lifecycle = lifecycleGeneration guard isCurrentLifecycleGeneration(lifecycle), isRegisteredInCatalog() else { return false } refreshGeneration &+= 1 @@ -301,8 +298,8 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { let incoming = CmuxTuiSnapshotParser.state(fromSnapshot: object, machine: machine) else { throw ProviderError.invalidSnapshot(machineID) } let installed = installSnapshotIfNewer(incoming, requestVersion: requestVersion) - // Equal cursors are a valid no-op refresh only when the accepted - // graph is byte-for-byte equivalent. A cursor alone is not proof + // Equal cursors are a valid no-op refresh only when the revisioned + // graph is equivalent. A cursor alone is not proof // that a malformed or misconfigured daemon returned the same graph. // A newer event can also win the race while this snapshot is in // flight; the final install-version check below covers that case. @@ -445,15 +442,16 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { return false } // A snapshot with the exact installed cursor is a valid no-op only when - // its graph and every pending receipt agree. This is important after a + // its revisioned graph and every pending receipt agree. This is important after a // rename: a delayed equal-cursor predecessor must not look current. if let current = cloudState, current.cursor == incoming.cursor { - guard current == incoming, incomingPassesPendingRenameFence(incoming) else { + guard current.hasSameRevisionedContent(as: incoming), incomingPassesPendingRenameFence(incoming) else { #if DEBUG cmuxDebugLog("cloud.state.snapshotIgnored machine=\(machineID) reason=equal-cursor-conflict") #endif return false } + cloudState = incoming retirePendingRemoteRenames(observed: incoming) return true } @@ -1187,7 +1185,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { if let starter = CmuxTuiSnapshotParser.createdTerminal(fromRunResult: object) { _ = recordCreatedTerminal(starter, workspaceID: id, name: nil, cwd: nil) } - _ = await refresh(force: true) + _ = await refreshCurrentGraph(force: true) return info.remoteWorkspaces?.first(where: { $0.id == id }) ?? provisional } @@ -1201,7 +1199,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { } // Validate against a fresh document. A cached workspace id can refer to // a closed or recycled daemon object after another client changes the VM. - guard await refresh(force: true), + guard await refreshCurrentGraph(force: true), let observed = cloudState, let previous = observed.workspaces.first(where: { $0.id == id }) else { throw SurfaceCatalogError.unsupported( @@ -1228,7 +1226,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // when this workspace still has the name we observed. If another // client changed it, do not overwrite that intent. guard Self.isRevisionConflict(error), - await refresh(force: true), + await refreshCurrentGraph(force: true), let latest = cloudState, let current = latest.workspaces.first(where: { $0.id == id }), let latestCursor = latest.cursor, @@ -1247,7 +1245,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // The command response is not the source of truth. Wait for the next // accepted snapshot/event so every local projection sees the same name. try Task.checkCancellation() - _ = await refresh(force: true) + _ = await refreshCurrentGraph(force: true) } /// Rename one placement-local daemon tab. This is the canonical path used by a @@ -1259,7 +1257,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // Creation and rename can arrive back-to-back. Refresh before validating the // target, but use the creation receipt when the accepted snapshot still // trails it. The receipt's revision is a CAS fence, not a timing guess. - let refreshEstablishedCurrentGraph = await refresh(force: true) + let refreshEstablishedCurrentGraph = await refreshCurrentGraph(force: true) try Task.checkCancellation() let pendingCreation = pendingCreation(forTabID: id) let pendingRename = pendingRemoteRename(for: .tab(id)) @@ -1296,7 +1294,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { recordPendingRename(tabID: id, name: normalizedName, revision: validated.revision) } catch { guard Self.isRevisionConflict(error), - await refresh(force: true), + await refreshCurrentGraph(force: true), let latest = cloudState, let current = latest.tabs.first(where: { $0.id == id }), let latestCursor = latest.cursor, @@ -1334,7 +1332,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // The daemon event normally installs this before the command exits. The // explicit read is the barrier for older clients that do not stream deltas. try Task.checkCancellation() - _ = await refresh(force: true) + _ = await refreshCurrentGraph(force: true) } /// Compatibility operation for callers that intentionally mean “all views”. @@ -1347,7 +1345,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // tab id, and use the fresh typed state for the old value. A creation // receipt supplies the one exact tab while the first snapshot catches // up, so an immediate rename cannot lose its target. - let refreshEstablishedCurrentGraph = await refresh(force: true) + let refreshEstablishedCurrentGraph = await refreshCurrentGraph(force: true) try Task.checkCancellation() let pending = pendingCreation(for: id) let observed = cloudState @@ -1437,7 +1435,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { // so a concurrent rename between checks cannot be overwritten. var compensated = renamedTabs.isEmpty if !renamedTabs.isEmpty, !mutationOutcomeUncertain, - await refresh(force: true), + await refreshCurrentGraph(force: true), let latest = cloudState, let latestCursor = latest.cursor, latestCursor.generation == observedCursor.generation, @@ -1464,7 +1462,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { } } } - _ = await refresh(force: true) + _ = await refreshCurrentGraph(force: true) if !compensated { throw Self.partialRenameError( id: id, @@ -1474,7 +1472,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { } throw error } - _ = await refresh(force: true) + _ = await refreshCurrentGraph(force: true) } @discardableResult @@ -2006,7 +2004,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { self.stateRecoveryRefreshTask = nil guard self.stateRecoveryRefreshQueued else { return } self.stateRecoveryRefreshQueued = false - await self.refresh(force: true) + await self.refreshCurrentGraph(force: true) if self.stateRecoveryRefreshQueued { self.scheduleStateRecoveryRefresh() } @@ -2030,7 +2028,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { guard !Task.isCancelled, let self else { return } self.scheduledRefresh = nil guard self.lifecycleGeneration == lifecycle, self.isRegisteredInCatalog() else { return } - await self.refresh(force: false) + await self.refreshCurrentGraph(force: false) } } @@ -2040,6 +2038,7 @@ final class CmuxTuiSurfaceProvider: SurfaceProvider { /// the placeholder. private func reprojectRestoredPanes(generation: UInt64) { guard isCurrentLifecycleGeneration(generation), isRegisteredInCatalog() else { return } + reprojectRestoredBrowserPanes(generation: generation) let terminals = catalog.snapshot.resources(on: machine).filter { $0.kind == .terminal } for terminal in terminals { for projection in catalog.projections(of: terminal.id) where !materializedPanels.contains(projection.panelID) { diff --git a/Sources/Surfaces/SurfaceCatalogModel.swift b/Sources/Surfaces/SurfaceCatalogModel.swift index 93bab5ea348b..4fc831f0d29a 100644 --- a/Sources/Surfaces/SurfaceCatalogModel.swift +++ b/Sources/Surfaces/SurfaceCatalogModel.swift @@ -960,6 +960,14 @@ struct CloudVMStateDocument: Hashable, Codable, Sendable { guard let data = Self.canonicalData(cursorObject) else { return false } values["cursor"] = data collections.removeValue(forKey: "cursor") + // session.revision mirrors the public cursor (resource_api.rs). Keep + // it aligned when a delta changes only resource rows. + if var session = value(forKey: "session") as? [String: Any], + let revision = session["revision"], CloudWireNumber.unsigned(revision) != nil { + session["revision"] = revision is String ? (String(cursor.revision) as Any) : NSNumber(value: cursor.revision) + guard let sessionData = Self.canonicalData(session) else { return false } + values["session"] = sessionData + } canonicalDataCache = nil return true } @@ -1269,19 +1277,6 @@ struct CloudVMState: Hashable, Codable, Sendable { // New archives contain one canonical document. The decoder keeps a // one-way rawSnapshot fallback for archives written before this model. - static func == (lhs: CloudVMState, rhs: CloudVMState) -> Bool { - lhs.machine == rhs.machine - && lhs.cursor == rhs.cursor - && lhs.document == rhs.document - && lhs.workspaces == rhs.workspaces - && lhs.screens == rhs.screens - && lhs.panes == rhs.panes - && lhs.tabs == rhs.tabs - && lhs.terminals == rhs.terminals - && lhs.browsers == rhs.browsers - && lhs.agents == rhs.agents - } - func hash(into hasher: inout Hasher) { hasher.combine(machine) hasher.combine(cursor) diff --git a/cmux.xcodeproj/project.pbxproj b/cmux.xcodeproj/project.pbxproj index 4e032919f490..20b75c2347b0 100644 --- a/cmux.xcodeproj/project.pbxproj +++ b/cmux.xcodeproj/project.pbxproj @@ -716,6 +716,8 @@ 7A0CE1000000000000000504 /* CloudPrivateNetworkPurpose.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A0CE1000000000000000503 /* CloudPrivateNetworkPurpose.swift */; }; 7A0CE1000000000000000304 /* CloudPrivateNetworkUse.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A0CE1000000000000000303 /* CloudPrivateNetworkUse.swift */; }; 0C45DC8D9BE54DDEA2397892 /* CloudPrivateRouteSelectionTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 1E6077A4B6CC441EB3668651 /* CloudPrivateRouteSelectionTests.swift */; }; + D1AC3F6282A94EAEB79D160B /* CloudProviderRefreshCoordinator.swift in Sources */ = {isa = PBXBuildFile; fileRef = A09B853868B54438A5B14F26 /* CloudProviderRefreshCoordinator.swift */; }; + AC38407F09E94DA38B94F168 /* CloudProviderRefreshCoordinatorTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 5C2A7143303749BE82E51F69 /* CloudProviderRefreshCoordinatorTests.swift */; }; 4DB42AF8827945DFADF7C4D4 /* CloudSidebarSurfaceRegressionTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F3E5958FB85E4B97957D4F73 /* CloudSidebarSurfaceRegressionTests.swift */; }; FA0000000000000000000005 /* CloudTerminalPaneClosure.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0000000000000000000006 /* CloudTerminalPaneClosure.swift */; }; FA0000000000000000000003 /* CloudTerminalPaneClosureTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0000000000000000000004 /* CloudTerminalPaneClosureTests.swift */; }; @@ -796,6 +798,8 @@ 7A0CE1000000000000000502 /* CloudTunnelTiming.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A0CE1000000000000000501 /* CloudTunnelTiming.swift */; }; C10D00010000000000000001 /* CloudVMActionLauncher.swift in Sources */ = {isa = PBXBuildFile; fileRef = C10D00020000000000000002 /* CloudVMActionLauncher.swift */; }; C75740010000000000000001 /* CloudVMMenuItemMetricsTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C75740010000000000000002 /* CloudVMMenuItemMetricsTests.swift */; }; + 2F48EE37A14B4D76A8013B83 /* CloudVMState+SnapshotComparison.swift in Sources */ = {isa = PBXBuildFile; fileRef = F18D73C012B643DBADE8B7B5 /* CloudVMState+SnapshotComparison.swift */; }; + 2D7F228001C24ABD865AFD66 /* CloudVMStateSnapshotComparisonTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7EFB9C9E7B5D4F39873871A1 /* CloudVMStateSnapshotComparisonTests.swift */; }; 999B711473384AA9593B1936 /* CloudWireGuardHub.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8F40FDCEAFAA24EB9A175A2 /* CloudWireGuardHub.swift */; }; 7A0CE1000000000000000710 /* CloudWireGuardHubDialer.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A0CE100000000000000070F /* CloudWireGuardHubDialer.swift */; }; 05FCD63D10A4F9C56B683B5A /* CloudWireGuardHubProcessSpawner.swift in Sources */ = {isa = PBXBuildFile; fileRef = 20C68DA526C0C2D39FE217B6 /* CloudWireGuardHubProcessSpawner.swift */; }; @@ -1130,6 +1134,7 @@ C11323210000000000000001 /* CmuxTuiSurfaceProvider+ManualMirror.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11323220000000000000001 /* CmuxTuiSurfaceProvider+ManualMirror.swift */; }; 779E7556FC96D977E6ECCFE3 /* CmuxTuiSurfaceProvider+PlacementSync.swift in Sources */ = {isa = PBXBuildFile; fileRef = 90C490DDBA08806E15C5CC0A /* CmuxTuiSurfaceProvider+PlacementSync.swift */; }; 7A0CE1000000000000000728 /* CmuxTuiSurfaceProvider+PortForward.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A0CE1000000000000000727 /* CmuxTuiSurfaceProvider+PortForward.swift */; }; + A4E17A019B254CA1AC30620A /* CmuxTuiSurfaceProvider+Refresh.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7289F02703E9435CBA1C598B /* CmuxTuiSurfaceProvider+Refresh.swift */; }; A1B2C3D4E5F6071827364961 /* CmuxTuiSurfaceProviderRegistry.swift in Sources */ = {isa = PBXBuildFile; fileRef = A1B2C3D4E5F6071827364960 /* CmuxTuiSurfaceProviderRegistry.swift */; }; D01100800000000000000008 /* CmuxTuiSurfaceProviderRegistryDiscoveryTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = D01100800000000000000007 /* CmuxTuiSurfaceProviderRegistryDiscoveryTests.swift */; }; 7A1D4C000000000000000106 /* CmuxTuiSurfaceProviderRegistryPollingTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A1D4C000000000000000105 /* CmuxTuiSurfaceProviderRegistryPollingTests.swift */; }; @@ -4254,6 +4259,8 @@ 7A0CE1000000000000000503 /* CloudPrivateNetworkPurpose.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudPrivateNetworkPurpose.swift; sourceTree = ""; }; 7A0CE1000000000000000303 /* CloudPrivateNetworkUse.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudPrivateNetworkUse.swift; sourceTree = ""; }; 1E6077A4B6CC441EB3668651 /* CloudPrivateRouteSelectionTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudPrivateRouteSelectionTests.swift; sourceTree = ""; }; + A09B853868B54438A5B14F26 /* CloudProviderRefreshCoordinator.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "CloudProviderRefreshCoordinator.swift"; sourceTree = ""; }; + 5C2A7143303749BE82E51F69 /* CloudProviderRefreshCoordinatorTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudProviderRefreshCoordinatorTests.swift; sourceTree = ""; }; F3E5958FB85E4B97957D4F73 /* CloudSidebarSurfaceRegressionTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudSidebarSurfaceRegressionTests.swift; sourceTree = ""; }; FA0000000000000000000006 /* CloudTerminalPaneClosure.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTerminalPaneClosure.swift; sourceTree = ""; }; FA0000000000000000000004 /* CloudTerminalPaneClosureTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTerminalPaneClosureTests.swift; sourceTree = ""; }; @@ -4331,6 +4338,8 @@ 7A0CE1000000000000000501 /* CloudTunnelTiming.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudTunnelTiming.swift; sourceTree = ""; }; C10D00020000000000000002 /* CloudVMActionLauncher.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudVMActionLauncher.swift; sourceTree = ""; }; C75740010000000000000002 /* CloudVMMenuItemMetricsTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudVMMenuItemMetricsTests.swift; sourceTree = ""; }; + F18D73C012B643DBADE8B7B5 /* CloudVMState+SnapshotComparison.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "CloudVMState+SnapshotComparison.swift"; sourceTree = ""; }; + 7EFB9C9E7B5D4F39873871A1 /* CloudVMStateSnapshotComparisonTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudVMStateSnapshotComparisonTests.swift; sourceTree = ""; }; D8F40FDCEAFAA24EB9A175A2 /* CloudWireGuardHub.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudWireGuardHub.swift; sourceTree = ""; }; 7A0CE100000000000000070F /* CloudWireGuardHubDialer.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudWireGuardHubDialer.swift; sourceTree = ""; }; 20C68DA526C0C2D39FE217B6 /* CloudWireGuardHubProcessSpawner.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudWireGuardHubProcessSpawner.swift; sourceTree = ""; }; @@ -4584,6 +4593,7 @@ C11323220000000000000001 /* CmuxTuiSurfaceProvider+ManualMirror.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "CmuxTuiSurfaceProvider+ManualMirror.swift"; sourceTree = ""; }; 90C490DDBA08806E15C5CC0A /* CmuxTuiSurfaceProvider+PlacementSync.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = "CmuxTuiSurfaceProvider+PlacementSync.swift"; sourceTree = ""; }; 7A0CE1000000000000000727 /* CmuxTuiSurfaceProvider+PortForward.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = "CmuxTuiSurfaceProvider+PortForward.swift"; sourceTree = ""; }; + 7289F02703E9435CBA1C598B /* CmuxTuiSurfaceProvider+Refresh.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "CmuxTuiSurfaceProvider+Refresh.swift"; sourceTree = ""; }; A1B2C3D4E5F6071827364960 /* CmuxTuiSurfaceProviderRegistry.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CmuxTuiSurfaceProviderRegistry.swift; sourceTree = ""; }; D01100800000000000000007 /* CmuxTuiSurfaceProviderRegistryDiscoveryTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CmuxTuiSurfaceProviderRegistryDiscoveryTests.swift; sourceTree = ""; }; 7A1D4C000000000000000105 /* CmuxTuiSurfaceProviderRegistryPollingTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CmuxTuiSurfaceProviderRegistryPollingTests.swift; sourceTree = ""; }; @@ -7294,6 +7304,9 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef 3349CDE9CB71FA353EA004D3 /* CmuxTuiSnapshotParser+Placement.swift */, AAED489C1C23533995AF55EA /* SurfaceProvider.swift */, F11347000000000000000002 /* SurfaceCatalog+CloudPorts.swift */, + A09B853868B54438A5B14F26 /* CloudProviderRefreshCoordinator.swift */, + 7289F02703E9435CBA1C598B /* CmuxTuiSurfaceProvider+Refresh.swift */, + F18D73C012B643DBADE8B7B5 /* CloudVMState+SnapshotComparison.swift */, D01100800000000000000001 /* SurfaceCatalogQueryService.swift */, 278BEFB3583AED9627966DFF /* SurfaceSocketCommands.swift */, 47BBA1A8703756FDDFD53CF3 /* Workspace+SurfaceCatalog.swift */, @@ -10067,6 +10080,8 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef 64C68B378B6071B213B4AA57 /* LocalSurfaceProviderTests.swift */, BC427A4C642EBC36E600FA7F /* CmuxTuiSurfaceProviderTests.swift */, 64CB7AF7545F56E95A91B154 /* CloudNotificationSyncTests.swift */, + 5C2A7143303749BE82E51F69 /* CloudProviderRefreshCoordinatorTests.swift */, + 7EFB9C9E7B5D4F39873871A1 /* CloudVMStateSnapshotComparisonTests.swift */, D01100800000000000000003 /* SurfaceCatalogQueryServiceTests.swift */, D01100800000000000000005 /* CloudCatalogQueryTestProvider.swift */, 877BC9A2B697FEF2957397FA /* SurfaceCatalogTests.swift */, @@ -11510,6 +11525,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef 7A0CE1000000000000000027 /* CloudPrivateNetworkGate.swift in Sources */, 7A0CE1000000000000000504 /* CloudPrivateNetworkPurpose.swift in Sources */, 7A0CE1000000000000000304 /* CloudPrivateNetworkUse.swift in Sources */, + D1AC3F6282A94EAEB79D160B /* CloudProviderRefreshCoordinator.swift in Sources */, FA0000000000000000000005 /* CloudTerminalPaneClosure.swift in Sources */, C986A0080000000000000001 /* CloudTreeActiveDrag.swift in Sources */, 3BCFA4F10BC6F6673CFA3BFC /* CloudTreeCellView.swift in Sources */, @@ -11573,6 +11589,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef 7A0CE1000000000000000722 /* CloudTunnelStatusModel.swift in Sources */, 7A0CE1000000000000000502 /* CloudTunnelTiming.swift in Sources */, C10D00010000000000000001 /* CloudVMActionLauncher.swift in Sources */, + 2F48EE37A14B4D76A8013B83 /* CloudVMState+SnapshotComparison.swift in Sources */, 999B711473384AA9593B1936 /* CloudWireGuardHub.swift in Sources */, 7A0CE1000000000000000710 /* CloudWireGuardHubDialer.swift in Sources */, 05FCD63D10A4F9C56B683B5A /* CloudWireGuardHubProcessSpawner.swift in Sources */, @@ -11654,6 +11671,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef C11323210000000000000001 /* CmuxTuiSurfaceProvider+ManualMirror.swift in Sources */, 779E7556FC96D977E6ECCFE3 /* CmuxTuiSurfaceProvider+PlacementSync.swift in Sources */, 7A0CE1000000000000000728 /* CmuxTuiSurfaceProvider+PortForward.swift in Sources */, + A4E17A019B254CA1AC30620A /* CmuxTuiSurfaceProvider+Refresh.swift in Sources */, A1B2C3D4E5F6071827364961 /* CmuxTuiSurfaceProviderRegistry.swift in Sources */, 036359ABDB1F58B489D378A4 /* CmuxTuiSurfaceProviders.swift in Sources */, 9520A0029520A0029520A002 /* CmuxVaultAgentRegistration+Hermes.swift in Sources */, @@ -13761,6 +13779,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef F11347000000000000000003 /* CloudPortOpenRegressionTests.swift in Sources */, 7A0CE100000000000000072C /* CloudPortRoutePlanTests.swift in Sources */, 0C45DC8D9BE54DDEA2397892 /* CloudPrivateRouteSelectionTests.swift in Sources */, + AC38407F09E94DA38B94F168 /* CloudProviderRefreshCoordinatorTests.swift in Sources */, 4DB42AF8827945DFADF7C4D4 /* CloudSidebarSurfaceRegressionTests.swift in Sources */, FA0000000000000000000003 /* CloudTerminalPaneClosureTests.swift in Sources */, B5C68F2FAFD21A103AFF429D /* CloudTreeMachineMenuTests.swift in Sources */, @@ -13775,6 +13794,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef 7A0CE100000000000000072E /* CloudTunnelStatusBlockerTests.swift in Sources */, 7A0CE1000000000000000402 /* CloudTunnelTestFakes.swift in Sources */, C75740010000000000000001 /* CloudVMMenuItemMetricsTests.swift in Sources */, + 2D7F228001C24ABD865AFD66 /* CloudVMStateSnapshotComparisonTests.swift in Sources */, 62AABF1720674956200EC388 /* CloudWireGuardHubTests.swift in Sources */, C0DE43000000000000000009 /* CmuxAgentChatConfigTests.swift in Sources */, 8295A0058295A0058295A005 /* CmuxAlertContentTests.swift in Sources */, diff --git a/cmuxTests/CloudProviderRefreshCoordinatorTests.swift b/cmuxTests/CloudProviderRefreshCoordinatorTests.swift new file mode 100644 index 000000000000..325cf859f410 --- /dev/null +++ b/cmuxTests/CloudProviderRefreshCoordinatorTests.swift @@ -0,0 +1,123 @@ +import Foundation +import Testing + +#if canImport(cmux_DEV) +@testable import cmux_DEV +#elseif canImport(cmux) +@testable import cmux +#endif + +@MainActor +struct CloudProviderRefreshCoordinatorTests { + @Test("Concurrent catalog reads cannot supersede the initial graph publication") + func concurrentReadersShareTheFirstPublication() async { + let coordinator = CloudProviderRefreshCoordinator() + let started = CloudLinkFirstValue() + let release = CloudLinkFirstValue() + let joining = CloudLinkFirstValue() + var generation = 0 + let operation: @MainActor (Bool) async -> Bool = { _ in + generation += 1 + let mine = generation + started.resolve(true) + _ = await release.result + return mine == generation + } + let first = Task { await coordinator.refresh(force: false, operation: operation) } + _ = await started.result + let second = Task { + joining.resolve(true) + return await coordinator.refresh(force: false, operation: operation) + } + _ = await joining.result + #expect(generation == 1) + release.resolve(true) + #expect(await first.value) + #expect(await second.value) + } + + @Test("Forced reads queued during a snapshot share one later forced pass") + func forcedReadersWaitForAReadStartedAfterTheirRequest() async { + let coordinator = CloudProviderRefreshCoordinator() + let started = CloudLinkFirstValue() + let release = CloudLinkFirstValue() + let firstWaiting = CloudLinkFirstValue() + let secondWaiting = CloudLinkFirstValue() + var forces: [Bool] = [] + let operation: @MainActor (Bool) async -> Bool = { force in + forces.append(force) + started.resolve(true) + _ = await release.result + return true + } + let background = Task { await coordinator.refresh(force: false, operation: operation) } + _ = await started.result + let first = Task { + firstWaiting.resolve(true) + return await coordinator.refresh(force: true, operation: operation) + } + _ = await firstWaiting.result + let second = Task { + secondWaiting.resolve(true) + return await coordinator.refresh(force: true, operation: operation) + } + _ = await secondWaiting.result + #expect(forces == [false]) + release.resolve(true) + #expect(await background.value) + #expect(await first.value) + #expect(await second.value) + #expect(forces == [false, true]) + } + + @Test("A metadata change restarts an invalidated pass before releasing its readers") + func invalidatedPassFinishesWithTheCurrentGraph() async { + let coordinator = CloudProviderRefreshCoordinator() + let started = CloudLinkFirstValue() + let release = CloudLinkFirstValue() + var calls = 0 + let read = Task { + await coordinator.refresh(force: true) { _ in + calls += 1 + if calls == 1 { + started.resolve(true) + _ = await release.result + return false + } + return true + } + } + _ = await started.result + coordinator.invalidate() + release.resolve(true) + #expect(await read.value) + #expect(calls == 2) + } + + @Test("Retiring a provider cancels its current pass and queued forced readers") + func cancellationRetiresQueuedRequests() async { + let coordinator = CloudProviderRefreshCoordinator() + let started = CloudLinkFirstValue() + let release = CloudLinkFirstValue() + let waiting = CloudLinkFirstValue() + var calls = 0 + let operation: @MainActor (Bool) async -> Bool = { _ in + calls += 1 + started.resolve(true) + _ = await release.result + return true + } + let first = Task { await coordinator.refresh(force: false, operation: operation) } + _ = await started.result + let forced = Task { + waiting.resolve(true) + return await coordinator.refresh(force: true, operation: operation) + } + _ = await waiting.result + coordinator.cancel() + release.resolve(true) + #expect(await first.value == false) + #expect(await forced.value == false) + #expect(calls == 1) + } +} diff --git a/cmuxTests/CloudVMStateSnapshotComparisonTests.swift b/cmuxTests/CloudVMStateSnapshotComparisonTests.swift new file mode 100644 index 000000000000..b9ca2c06f068 --- /dev/null +++ b/cmuxTests/CloudVMStateSnapshotComparisonTests.swift @@ -0,0 +1,97 @@ +import Foundation +import Testing + +#if canImport(cmux_DEV) +@testable import cmux_DEV +#elseif canImport(cmux) +@testable import cmux +#endif + +struct CloudVMStateSnapshotComparisonTests { + private func snapshot() -> [String: Any] { + [ + "cursor": ["generation": "daemon-1", "revision": "2"], + "workspaces": [["id": "ws-1", "name": "Original", "focused": true]], + "screens": [], "panes": [], "tabs": [], "terminals": [], "browsers": [], "agents": [], + "clients": [["id": "client-1", "connected_seconds": 1]] + ] + } + + private func state(_ object: [String: Any]) throws -> CloudVMState { + try #require(CmuxTuiSnapshotParser.state(fromSnapshot: object, machine: .cloud("vm-test"))) + } + + @Test("Connection age and request-client churn do not invalidate a session revision") + func volatileClientsDoNotMakeAnUnchangedGraphStale() throws { + let before = try state(snapshot()) + var changed = snapshot() + changed["clients"] = [ + ["id": "client-1", "connected_seconds": 45], + ["id": "snapshot-reader", "connected_seconds": 0] + ] + let after = try state(changed) + + #expect(before != after, "Diagnostics must remain in the complete exported document") + #expect(before.hasSameRevisionedContent(as: after)) + } + + @Test("Live terminal geometry can change without changing the resource revision") + func terminalResizeDoesNotInvalidateTheGraph() throws { + var object = snapshot() + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "cols": 80, "rows": 24]] + let before = try state(object) + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "cols": 120, "rows": 40]] + let after = try state(object) + #expect(before != after) + #expect(before.hasSameRevisionedContent(as: after)) + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "cols": 120, "rows": 40, "future_field": "changed"]] + #expect(try !before.hasSameRevisionedContent(as: state(object))) + } + + @Test("PTY title updates remain live observations while launch identity stays strict") + func terminalTitleDoesNotInvalidateTheGraph() throws { + var object = snapshot() + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "title": "bash", "cwd": "/home/cmux"]] + let before = try state(object) + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "title": "vim", "cwd": "/home/cmux"]] + let after = try state(object) + #expect(before != after) + #expect(before.hasSameRevisionedContent(as: after)) + object["terminals"] = [["id": "term-1", "running": true, "lifecycle": "running", "title": "vim", "cwd": "/different-launch"]] + #expect(try !before.hasSameRevisionedContent(as: state(object))) + } + + @Test("Actual same-cursor conflicts remain rejected", arguments: ["workspaces", "terminals", "future_resources", "cursor"]) + func graphChangesRemainConflicts(field: String) throws { + let before = try state(snapshot()) + var changed = snapshot() + switch field { + case "workspaces": changed[field] = [["id": "ws-1", "name": "Changed", "focused": true]] + case "terminals": changed[field] = [["id": "term-new", "running": true, "lifecycle": "running"]] + case "cursor": changed[field] = ["generation": "daemon-1", "revision": "3"] + default: changed[field] = [["id": "future-1", "value": "changed"]] + } + #expect(try !before.hasSameRevisionedContent(as: state(changed))) + } + + @Test("Applying a delta keeps the session revision aligned with its cursor", arguments: [false, true]) + func deltaAdvancesBothRevisionRepresentations(legacyNumeric: Bool) throws { + var object = snapshot() + object["session"] = ["id": "session-1", "revision": legacyNumeric ? (2 as Any) : "2", "name": "Kept"] + var document = CloudVMStateDocument(snapshot: object) + let advanced = document.setCursor(CloudVMCursor(generation: "daemon-1", revision: 3)) + #expect(advanced) + let session = try #require(document.value(forKey: "session") as? [String: Any]) + #expect(CloudWireNumber.unsigned(session["revision"]) == 3) + #expect(session["name"] as? String == "Kept") + #expect((session["revision"] is String) == !legacyNumeric) + } + + @Test("A legacy document without a session object does not gain one") + func deltaDoesNotInventASessionRecord() { + var document = CloudVMStateDocument(snapshot: snapshot()) + let advanced = document.setCursor(CloudVMCursor(generation: "daemon-1", revision: 3)) + #expect(advanced) + #expect(document.value(forKey: "session") == nil) + } +} diff --git a/web/services/vms/drivers/freestyle.ts b/web/services/vms/drivers/freestyle.ts index a3d2c56db921..a47b97924e79 100644 --- a/web/services/vms/drivers/freestyle.ts +++ b/web/services/vms/drivers/freestyle.ts @@ -9,6 +9,8 @@ import { type VpcData, } from "freestyle"; import { randomBytes } from "node:crypto"; +import { Effect } from "effect"; +import { announceFreestyleNetwork } from "./freestyleNetworkAnnouncement"; import { ProviderError, type AttachEndpoint, @@ -260,17 +262,10 @@ type FreestyleNetworkAddress = { * degraded path but a guaranteed timeout with a misleading address in the * error. * - * Within the network, IPv4 is preferred, because only the v4 path is reliable - * over the WireGuard tunnel. The tunnel routes the VPC's v4 prefix as a subnet, - * so it reaches any member the moment that member exists; its v6 path does not - * pick up members created after the tunnel came up. A VM created into an - * established tunnel therefore answers on its private v4 and blackholes on its - * private v6 from the same Mac, while both work VM-to-VM inside the VPC. With - * v6 first, every freshly created machine spent the full 60s connect timeout - * and surfaced as "Command timed out"; only machines predating the tunnel - * connected. Preferring v4 also matches the app's own `preferredPrivateAddress` - * (v4 then v6), so the address a person copies from the sidebar is the address - * the daemon is dialed on. + * Within the network, IPv4 is the legacy preferred route. Clients also receive + * both private addresses and select the reachable family through their hub. + * Announcing the guest's assigned addresses prepares the provider's forwarding + * state; either family can independently be unavailable during a network fault. */ export function freestyleCmuxRemoteRoute(addresses: FreestyleRouteAddresses, vmId: string): string { const networks = addresses.vpcs ?? addresses.networks ?? []; @@ -1001,6 +996,7 @@ export class FreestyleProvider implements VMProvider { } // The baked supervisor is already bringing the daemon up; the only // per-machine input it needs is the model-plane env file. + await this.announcePrivateAddresses(vm, data); } catch (err) { // A VM that failed to size or configure must not survive as an // orphan, and an undersized machine must not ship as if it were @@ -1118,6 +1114,18 @@ export class FreestyleProvider implements VMProvider { const fs = this.deps.client(CREATE_TIMEOUT_MS); const vm = fs.vms.ref(vmId); const data = await vm.start(); + // The wake already succeeded: the machine is running and, unlike + // create and restore, there is no fresh allocation to roll back. A + // start payload can also name the network a VM is on before the + // platform fills in the address assigned on it, so an unusable + // address here is not a verdict on the machine. Prepare the route + // best-effort and leave readiness to openCmuxRemote, which reads the + // authoritative addresses and fails closed on them. + try { + await this.announcePrivateAddresses(vm, data); + } catch (announceError) { + recordSpanError(span, announceError); + } // Older cmux machines were created while the provider's account // default supplied a finite idle timeout. Clear that legacy policy // the first time the user wakes one so the box stays available after @@ -1267,6 +1275,12 @@ export class FreestyleProvider implements VMProvider { // are mandatory: a snapshot never carries a token, so the restored // machine is unusable until its own injection is live. await this.ensureCmuxTuiRunning(vm, vmId).catch(() => undefined); + try { + await this.announcePrivateAddresses(vm, data); + } catch (error) { + await vm.delete().catch((cleanupError) => recordSpanError(span, cleanupError)); + throw error; + } return { provider: "freestyle" as const, providerVmId: vmId, @@ -1297,6 +1311,7 @@ export class FreestyleProvider implements VMProvider { const persisted = freestyleRouteAddressesFromMetadata(options?.providerMetadata); const data = persisted ?? await vm.data(); const route = freestyleCmuxRemoteRoute(data, vmId); + await this.announcePrivateAddresses(vm, data); const fingerprint = options?.deviceFingerprint; let { bundle, healed } = await this.loadCmuxRemoteBundle(vm, vmId, fingerprint); // A daemon from a bake that predates the trusted listener is brought to @@ -1402,6 +1417,14 @@ export class FreestyleProvider implements VMProvider { return Object.keys(networkAddresses).length ? { networkAddresses } : {}; } + private async announcePrivateAddresses(vm: Vm, data: FreestyleRouteAddresses): Promise { + const addresses = (data.vpcs ?? data.networks ?? []) + .flatMap((network) => [network.ipv4, network.ipv6]) + .filter((address): address is string => typeof address === "string" && address.trim() !== "") + .map((address) => address.trim()); + await Effect.runPromise(announceFreestyleNetwork(vm, addresses)); + } + async approveCmuxRemoteEnrollment( vmId: string, invitationId: string, diff --git a/web/services/vms/drivers/freestyleNetworkAnnouncement.ts b/web/services/vms/drivers/freestyleNetworkAnnouncement.ts new file mode 100644 index 000000000000..a38a74618cb1 --- /dev/null +++ b/web/services/vms/drivers/freestyleNetworkAnnouncement.ts @@ -0,0 +1,68 @@ +import { Effect } from "effect"; +import type { Vm } from "freestyle"; +import { isIP } from "node:net"; +import { shellQuote } from "./cmuxTuiDaemon"; +import { ProviderError } from "./types"; + +/** + * A resumed snapshot can acquire a VPC address before the provider learns its + * link-layer mapping. Announce only addresses actually assigned to this guest: + * one gratuitous ARP for IPv4 and one unsolicited neighbor advertisement for + * IPv6. One available family is sufficient; clients retain their address race. + * No routes, firewall rules, interfaces, or running sessions are changed. + */ +export function freestyleNetworkAnnouncementCommand(addresses: readonly string[]): string { + const script = `import ipaddress,json,socket,struct,subprocess,sys +expected = {ipaddress.ip_address(value) for value in json.loads(sys.argv[1])} +links = json.loads(subprocess.check_output(['ip', '-j', 'address', 'show'], timeout=3)) +announced = set() +for link in links: + if link.get('link_type') != 'ether' or 'UP' not in link.get('flags', []): + continue + mac = bytes.fromhex(link['address'].replace(':', '')) + for address in link.get('addr_info', []): + ip = ipaddress.ip_address(address['local']) + if ip not in expected or ip in announced: + continue + try: + if ip.version == 4: + packet = b'\\xff'*6 + mac + struct.pack('!HHHBBH', 0x0806, 1, 0x0800, 6, 4, 1) + packet += mac + ip.packed + b'\\x00'*6 + ip.packed + with socket.socket(socket.AF_PACKET, socket.SOCK_RAW) as stream: + stream.bind((link['ifname'], 0)) + stream.send(packet) + else: + index = link['ifindex'] + packet = struct.pack('!BBHI', 136, 0, 0, 0x20000000) + ip.packed + bytes([2, 1]) + mac + with socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_ICMPV6) as stream: + stream.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, index) + stream.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 255) + stream.bind((str(ip), 0, 0, index)) + stream.sendto(packet, ('ff02::1', 0, 0, index)) + except OSError: + continue + announced.add(ip) +if not announced: + raise SystemExit('Private network addresses are not ready on the guest') +`; + return `python3 -c ${shellQuote(script)} ${shellQuote(JSON.stringify(addresses))}`; +} + +/** + * The shared private-address setup every lifecycle path runs. It fails closed: + * a machine with no usable address has no route to its daemon or its ports. + * Create, restore, and attach surface that failure; a wake reports it without + * failing, having nothing to roll back (see FreestyleProvider.resume). + */ +export function announceFreestyleNetwork(vm: Pick, addresses: readonly string[]) { + const valid = [...new Set(addresses.filter((address) => isIP(address) !== 0))]; + if (valid.length === 0) { + return Effect.fail(new ProviderError("freestyle", "Private network has no valid assigned address")); + } + return Effect.tryPromise({ + try: () => vm.exec({ command: freestyleNetworkAnnouncementCommand(valid), linuxUser: "root", timeoutMs: 5_000 }), + catch: (cause) => new ProviderError("freestyle", "announce private network", cause), + }).pipe(Effect.flatMap((result) => result.statusCode === 0 + ? Effect.void + : Effect.fail(new ProviderError("freestyle", "Private network announcement failed", result.stderr)))); +} diff --git a/web/tests/freestyle-network-announcement.test.ts b/web/tests/freestyle-network-announcement.test.ts index 37a32e9b0c36..0258ed201ac7 100644 --- a/web/tests/freestyle-network-announcement.test.ts +++ b/web/tests/freestyle-network-announcement.test.ts @@ -1,17 +1,66 @@ import { describe, expect, test } from "bun:test"; import type { Freestyle } from "freestyle"; +import { Effect } from "effect"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { spawnSync } from "node:child_process"; import { FreestyleProvider } from "../services/vms/drivers/freestyle"; +import { announceFreestyleNetwork, freestyleNetworkAnnouncementCommand } from "../services/vms/drivers/freestyleNetworkAnnouncement"; + +function captureAnnouncements(addresses: string[], failingFamily = "") { + const directory = mkdtempSync(join(tmpdir(), "cmux-network-test-")); + const capture = join(directory, "packets.json"); + // Execute the shipped guest command with only its OS boundary substituted. + // No raw socket, subprocess, or host network operation can escape this fixture. + writeFileSync(join(directory, "sitecustomize.py"), `import atexit,json,os,socket,subprocess +packets=[] +class Socket: + def __init__(self,*args): self.bound=None; self.options=[] + def __enter__(self): return self + def __exit__(self,*args): pass + def bind(self,value): self.bound=value + def setsockopt(self,*args): self.options.append(args) + def send(self,packet): + if os.environ['FAILING_FAMILY'] in ['ipv4','both']: raise OSError('IPv4 unavailable') + packets.append(dict(bound=self.bound,packet=packet.hex(),options=self.options)) + def sendto(self,packet,target): + if os.environ['FAILING_FAMILY'] in ['ipv6','both']: raise OSError('IPv6 unavailable') + packets.append(dict(bound=self.bound,packet=packet.hex(),target=target,options=self.options)) +socket.socket=Socket +socket.AF_PACKET=17 +links=[dict(ifname='eth0.181',ifindex=8,link_type='ether',flags=['UP'],address='02:00:0a:10:00:02',addr_info=[dict(local='10.16.0.2'),dict(local='fd00::2'),dict(local='fe80::2')])] +subprocess.check_output=lambda *args,**kwargs: json.dumps(links).encode() +atexit.register(lambda: open(os.environ['CAPTURE_PATH'],'w').write(json.dumps(packets))) +`); + try { + const result = spawnSync("/bin/sh", ["-c", freestyleNetworkAnnouncementCommand(addresses)], { + env: { ...process.env, PYTHONPATH: directory, CAPTURE_PATH: capture, FAILING_FAMILY: failingFamily }, encoding: "utf8", + }); + return { status: result.status, packets: JSON.parse(readFileSync(capture, "utf8")) as Array<{ + bound: Array; packet: string; target?: Array; options: number[][]; + }> }; + } finally { rmSync(directory, { recursive: true, force: true }); } +} describe("Freestyle private network readiness", () => { - test("create prepares the guest network before publishing its private addresses", async () => { + test.each([ + { operation: "create", hasAddresses: true }, + { operation: "create", hasAddresses: false }, + { operation: "restore", hasAddresses: true }, + { operation: "restore", hasAddresses: false }, + ])("publishes only after network readiness or rolls back: %j", async ({ operation, hasAddresses }) => { const events: string[] = []; const data = { id: "vm-network-test", state: "running", snapshotId: "sh-fixture", resources: { cpu: 64, memory: 131072, storage: 1048576 }, - vpcs: [{ ipv4: "10.16.0.2", ipv6: "fd00::2" }], + vpcs: hasAddresses ? [{ ipv4: "10.16.0.2", ipv6: "fd00::2" }] : [], }; const vm = { - exec: async () => { events.push("guest-network"); return { statusCode: 0, stdout: "", stderr: "" }; }, + exec: async ({ command }: { command: string }) => { + events.push(command.startsWith("python3 -c ") ? "guest-network" : "guest-daemon"); + return { statusCode: 0, stdout: "", stderr: "" }; + }, delete: async () => { events.push("delete"); }, }; const client = { vms: { @@ -23,9 +72,75 @@ describe("Freestyle private network readiness", () => { resolveDaemonSource: async () => { throw new Error("No daemon install is needed"); }, }); - await provider.create({ image: "sh-fixture", network: { id: "vpc-fixture" } }); - events.push("published"); + const allocation = operation === "create" + ? provider.create({ image: "sh-fixture", network: { id: "vpc-fixture" } }) + : provider.restore("sh-fixture", { network: { id: "vpc-fixture" } }); + const preparation = operation === "restore" ? ["allocated", "guest-daemon"] : ["allocated"]; + if (hasAddresses) { + await allocation; + events.push("published"); + expect(events).toEqual([...preparation, "guest-network", "published"]); + } else { + await expect(allocation).rejects.toThrow(); + expect(events).toEqual([...preparation, "delete"]); + } + }); + + test("the guest announces assigned IPv4 and IPv6 without touching other addresses", () => { + const { status, packets } = captureAnnouncements(["10.16.0.2", "fd00::2"]); + expect(status).toBe(0); + expect(packets).toHaveLength(2); + const arp = Buffer.from(packets[0].packet, "hex"); + expect(packets[0].bound).toEqual(["eth0.181", 0]); + expect(arp.subarray(0, 6).toString("hex")).toBe("ffffffffffff"); + expect(arp.readUInt16BE(12)).toBe(0x0806); + expect(arp.readUInt16BE(20)).toBe(1); + expect([...arp.subarray(28, 32)]).toEqual([10, 16, 0, 2]); + expect(arp.subarray(38, 42)).toEqual(arp.subarray(28, 32)); + const neighbor = Buffer.from(packets[1].packet, "hex"); + expect(neighbor[0]).toBe(136); + expect(neighbor.readUInt32BE(4)).toBe(0x20000000); + expect(neighbor.subarray(8, 24).toString("hex")).toBe("fd000000000000000000000000000002"); + expect(neighbor.subarray(24).toString("hex")).toBe("020102000a100002"); + expect(packets[1].target).toEqual(["ff02::1", 0, 0, 8]); + expect(packets[1].options.some((option) => option[2] === 255)).toBe(true); + }); + + test("an address absent from the guest is never advertised and fails readiness", () => { + const { status, packets } = captureAnnouncements(["10.16.0.99"]); + expect(status).not.toBe(0); + expect(packets).toEqual([]); + }); + + test("one assigned family remains usable while the other address is still pending", () => { + const { status, packets } = captureAnnouncements(["10.16.0.2", "fd00::99"]); + expect(status).toBe(0); + expect(packets).toHaveLength(1); + expect(Buffer.from(packets[0].packet, "hex").readUInt16BE(12)).toBe(0x0806); + }); + + test.each(["ipv4", "ipv6"])("an unavailable %s socket preserves the working family", (family) => { + const { status, packets } = captureAnnouncements(["10.16.0.2", "fd00::2"], family); + expect(status).toBe(0); + expect(packets).toHaveLength(1); + expect(packets[0].target !== undefined).toBe(family === "ipv4"); + }); + + test("failure of both families still rejects network readiness", () => { + const { status, packets } = captureAnnouncements(["10.16.0.2", "fd00::2"], "both"); + expect(status).not.toBe(0); + expect(packets).toEqual([]); + }); + + test("a guest failure prevents reporting that its network is ready", async () => { + const vm = { exec: async () => ({ statusCode: 1, stdout: "", stderr: "not assigned" }) }; + await expect(Effect.runPromise(announceFreestyleNetwork(vm as never, ["10.16.0.2"]))).rejects.toThrow(); + }); - expect(events).toEqual(["allocated", "guest-network", "published"]); + test.each([{ addresses: [] }, { addresses: ["invalid-address"] }])("missing usable addresses fail before guest execution: %j", async ({ addresses }) => { + let executed = false; + const vm = { exec: async () => { executed = true; return { statusCode: 0, stdout: "", stderr: "" }; } }; + await expect(Effect.runPromise(announceFreestyleNetwork(vm as never, addresses))).rejects.toThrow("Private network has no valid assigned address"); + expect(executed).toBe(false); }); }); diff --git a/web/tests/vm-devbox-desktop.test.ts b/web/tests/vm-devbox-desktop.test.ts index d4a35475b6ea..30c8d24f0cf3 100644 --- a/web/tests/vm-devbox-desktop.test.ts +++ b/web/tests/vm-devbox-desktop.test.ts @@ -141,8 +141,8 @@ describe("devbox desktop layer", () => { expect(startVnc).toContain(`-rfbport ${DEVBOX_DESKTOP_RFB_PORT}`); expect(startVnc).toContain("-SecurityTypes None"); expect(startVnc).toContain("-localhost"); - // The app's desktop port: the noVNC web client must answer on 6901. - expect(startVnc).toContain(`websockify --web /usr/share/novnc --heartbeat 30 0.0.0.0:${DEVBOX_DESKTOP_NOVNC_PORT} 127.0.0.1:${DEVBOX_DESKTOP_RFB_PORT}`); + // Listener reachability is exercised by verify-devbox-image over both + // families; do not pin the implementation to the old IPv4-only command. expect(dockerfile).toContain("ln -s vnc.html /usr/share/novnc/index.html"); expect(freestyleBake).toContain("ln -s vnc.html /usr/share/novnc/index.html"); // The verifier proves both ports from inside the VM (/proc/net/tcp, hex diff --git a/web/tests/vm-freestyle-provider.test.ts b/web/tests/vm-freestyle-provider.test.ts index b58b628924cd..45f63a9ea3b3 100644 --- a/web/tests/vm-freestyle-provider.test.ts +++ b/web/tests/vm-freestyle-provider.test.ts @@ -212,6 +212,7 @@ describe("Freestyle tunnel create recovery", () => { // delete so the driver's guest-facing behavior can be asserted without a // platform. `probeExit` is what the edge readiness probe returns. function fakeFreestyle(input: { readonly probeExit: number }) { + const networkData = { publicIpv6: "2602:f75c:0:1::2a", vpcs: [{ ipv4: "10.4.0.7", ipv6: "fd00:4::7" }] }; const creates: unknown[] = []; const resizes: unknown[] = []; const execs: string[] = []; @@ -231,7 +232,7 @@ function fakeFreestyle(input: { readonly probeExit: number }) { delete: async () => { deletes.push(VM_ID); }, - data: async () => ({ publicIpv6: "2602:f75c:0:1::2a" }), + data: async () => networkData, // Every VM boots at its snapshot's resources; create grows it to the plan // machine before bootstrap (see growToRequestedSize). resize: async (options: unknown) => { @@ -242,7 +243,7 @@ function fakeFreestyle(input: { readonly probeExit: number }) { vms: { create: async (options: unknown) => { creates.push(options); - return { vm, vmId: VM_ID, data: { publicIpv6: "2602:f75c:0:1::2a", vpcs: [] } }; + return { vm, vmId: VM_ID, data: networkData }; }, get: async () => ({ resources: { cpu: 2, memory: 4096, storage: 16384 } }), ref: () => vm, @@ -251,7 +252,7 @@ function fakeFreestyle(input: { readonly probeExit: number }) { return { client, creates, resizes, execs, writes, deletes }; } -function providerWith(fake: ReturnType): FreestyleProvider { +function providerWith(fake: { readonly client: Freestyle }): FreestyleProvider { return new FreestyleProvider({ client: () => fake.client, resolveDaemonSource: async () => ({ @@ -612,7 +613,11 @@ describe("FreestyleProvider create with edge rules", () => { vpcs: [{ vpcId: "vpc_1", ipv4: true, ipv6: true }], tls: { rules: freestyleEdgeRules([EDGE_RULE]) }, }); - expect(handle.providerMetadata).toEqual({ networkId: "vpc_1" }); + expect(handle.providerMetadata).toEqual({ + networkId: "vpc_1", + networkIpv4: "10.4.0.7", + networkIpv6: "fd00:4::7", + }); expect(JSON.stringify(fake.writes)).not.toContain("crt_"); }); @@ -655,6 +660,76 @@ describe("FreestyleProvider create with edge rules", () => { }); }); +// A wake: `start()` reports the machine running, and its payload may or may not +// carry the address the platform assigned it on the private network. `delete` +// and `pause` are recorded so a test can prove the wake rolled nothing back. +function resumeFake(vpcs?: readonly Record[]) { + const execs: string[] = []; + const deletes: string[] = []; + const pauses: string[] = []; + const vm = { + start: async () => ({ + id: VM_ID, + state: "running" as const, + snapshotId: "sh-devbox", + resources: { cpu: 2, memory: 4096, storage: 16384 }, + ...(vpcs === undefined ? {} : { vpcs }), + }), + update: async () => ({}), + exec: async ({ command }: { command: string }) => { + execs.push(command); + return { statusCode: 0, stdout: "", stderr: "" }; + }, + delete: async () => { + deletes.push(VM_ID); + }, + pause: async () => { + pauses.push(VM_ID); + }, + }; + const client = { vms: { ref: () => vm } } as unknown as Freestyle; + return { client, execs, deletes, pauses }; +} + +/** The addresses the guest announcement was asked to announce, if it ran. */ +function announcedAddresses(execs: readonly string[]): readonly string[] { + const announcement = execs.find((command) => command.includes("Private network addresses are not ready")); + const payload = announcement?.match(/'(\[[^\[\]]*\])'$/)?.[1]; + return payload ? (JSON.parse(payload) as string[]) : []; +} + +describe("FreestyleProvider resume network readiness", () => { + test("a wake announces the private addresses its payload carries", async () => { + const fake = resumeFake([{ vpcId: "vpc_1", ipv4: "10.4.0.7", ipv6: "fd00:4::7" }]); + + const handle = await providerWith(fake).resume(VM_ID); + + expect(handle.status).toBe("running"); + expect(announcedAddresses(fake.execs)).toEqual(["10.4.0.7", "fd00:4::7"]); + }); + + test.each([ + { vpcs: undefined }, + { vpcs: [] }, + { vpcs: [{ vpcId: "vpc_1", routes: [] }] }, + { vpcs: [{ ipv4: "not-an-ip", ipv6: "also-not-an-ip" }] }, + ])("a wake without a usable address still wakes and rolls nothing back: %j", async ({ vpcs }) => { + // `start()` has already returned, so the machine is running and a resume + // has no fresh allocation to undo. A start payload can also name the + // network before the platform fills in the address assigned on it, so a + // missing address here is not a verdict on the machine. openCmuxRemote + // reads the authoritative addresses and is the boundary that fails closed. + const fake = resumeFake(vpcs); + + const handle = await providerWith(fake).resume(VM_ID); + + expect(handle.status).toBe("running"); + expect(announcedAddresses(fake.execs)).toEqual([]); + expect(fake.deletes).toEqual([]); + expect(fake.pauses).toEqual([]); + }); +}); + describe("FreestyleProvider resume policy", () => { test("clears a legacy idle timeout when waking an existing machine", async () => { const updates: unknown[] = []; @@ -665,6 +740,7 @@ describe("FreestyleProvider resume policy", () => { snapshotId: "sh-devbox", resources: { cpu: 2, memory: 4096, storage: 16384 }, idleTimeoutSeconds: 3600, + vpcs: [{ ipv4: "10.4.0.7", ipv6: "fd00:4::7" }], }), update: async (options: unknown) => { updates.push(options);