Fix cloud terminal reconnect stalls during output bursts

Sources/Cloud/CloudTuiManualIOConnection.swift

@@ -8,29 +8,37 @@ import Foundation
 /// utility queue so Ghostty's main actor and input path never wait on a file
 /// descriptor or parse a large output burst.
 // @unchecked Sendable is safe here because every mutable descriptor/source/
-// framing field is accessed only on `queue`; the AsyncStream continuation is
-// the sole cross-thread handoff and carries immutable `Data` values.
+// framing field and pending demand are accessed only on `queue`. Continuations
+// hand immutable frames back to the single async consumer.
 final class CloudTuiManualIOConnection: @unchecked Sendable {
     private static let maximumLineBytes = 16 * 1024 * 1024
-    // One frame can be a protocol-sized replay. Keep the stream's retained
-    // payload bounded to four such frames; a fifth frame closes the attachment
-    // and lets the owner reconnect from a fresh snapshot instead of growing
-    // memory while the main actor is stalled.
-    private static let maximumBufferedFrames = 4
+    private static let readChunkBytes = 16 * 1024
 
     private let socketPath: String
     private let queue: DispatchQueue
     private let commandBuilder: CloudTuiManualIOCommand
-    let events: AsyncStream<CloudTuiManualIOFrame>
-    private let eventsContinuation: AsyncStream<CloudTuiManualIOFrame>.Continuation
+
+    /// Single-consumer stream. Each next() requests one frame, so a busy renderer
+    /// applies socket backpressure instead of overflowing a decoded-frame queue.
+    var events: AsyncStream<CloudTuiManualIOFrame> {
+        AsyncStream(unfolding: { [weak self] in
+            await self?.nextFrame()
+        }, onCancel: { [weak self] in
+            self?.close()
+        })
+    }
+
+    private var nextFrameContinuation: CheckedContinuation<CloudTuiManualIOFrame?, Never>?
     private var descriptor: Int32 = -1
     private var isConnected = false
     private var readSource: DispatchSourceRead?
+    private var readSourceSuspended = false
     private var writeSource: DispatchSourceWrite?
     private var writeSourceSuspended = true
     private var descriptorLease: CloudTuiManualIODescriptorLease?
     private var startContinuation: CheckedContinuation<Void, Error>?
     private var pendingLine = Data()
+    private var pendingLineSearchOffset = 0
     private var pendingWrites: [Data] = []
     private var pendingWriteOffset = 0
     private var pendingWriteBytes = 0
@@ -48,15 +56,25 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         self.socketPath = socketPath
         self.queue = queue
         self.commandBuilder = commandBuilder
-        // A stalled Ghostty parser must not let a remote output burst grow an
-        // unbounded in-memory queue. Dropping a frame would corrupt the VT
-        // stream, so the bounded overflow edge closes this attachment and lets
-        // the owner reconnect from a fresh snapshot.
-        (events, eventsContinuation) = AsyncStream<CloudTuiManualIOFrame>.makeStream(
-            bufferingPolicy: .bufferingOldest(Self.maximumBufferedFrames)
-        )
-        eventsContinuation.onTermination = { [weak self] _ in
-            self?.close()
+    }
+
+    private func nextFrame() async -> CloudTuiManualIOFrame? {
+        guard !Task.isCancelled else {
+            close()
+            return nil
+        }
+        return await withCheckedContinuation { continuation in
+            queue.async { [self] in
+                guard !closed, nextFrameContinuation == nil else {
+                    continuation.resume(returning: nil)
+                    return
+                }
+                nextFrameContinuation = continuation
+                resumeReadSourceLocked()
+                // A previous read may already contain the next complete line;
+                // do not depend on another socket-readability notification.
+                readAvailableLocked()
+            }
         }
     }
 
@@ -169,6 +187,9 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         // descriptor and close it exactly once.
         source.activate()
         writeSource.activate()
+        if nextFrameContinuation == nil {
+            suspendReadSourceLocked()
+        }
 
         var address = try Self.unixAddress(path: socketPath)
         let addressLength = socklen_t(Self.unixAddressLength(address: address))
@@ -214,6 +235,7 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         let continuation = startContinuation
         startContinuation = nil
         continuation?.resume()
+        readAvailableLocked()
     }
 
     private func failStartLocked(_ error: Error) {
@@ -224,35 +246,39 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
     }
 
     private func readAvailableLocked() {
-        guard !closed, isConnected, descriptor >= 0 else { return }
-        var bytes = [UInt8](repeating: 0, count: 16 * 1024)
+        guard !closed, isConnected, descriptor >= 0, nextFrameContinuation != nil else { return }
+        var bytes = [UInt8](repeating: 0, count: Self.readChunkBytes)
         while !closed {
-            let count = Darwin.read(descriptor, &bytes, bytes.count)
-            if count > 0 {
-                pendingLine.append(bytes, count: count)
-                guard pendingLine.count <= Self.maximumLineBytes else {
+            while let newline = pendingLine[
+                pendingLine.index(pendingLine.startIndex, offsetBy: pendingLineSearchOffset)...
+            ].firstIndex(of: 0x0A) {
+                guard pendingLine.distance(from: pendingLine.startIndex, to: newline) <= Self.maximumLineBytes else {
                     closeLocked()
                     return
                 }
-                while let newline = pendingLine.firstIndex(of: 0x0A) {
-                    let line = Data(pendingLine[..<newline])
-                    pendingLine.removeSubrange(...newline)
-                    if !line.isEmpty {
-                        guard let frame = CloudTuiManualIOFrameDecoder().decode(line) else {
-                            continue
-                        }
-                        switch eventsContinuation.yield(frame) {
-                        case .enqueued:
-                            break
-                        case .dropped, .terminated:
-                            closeLocked()
-                            return
-                        @unknown default:
-                            closeLocked()
-                            return
-                        }
-                    }
-                }
+                let line = Data(pendingLine[..<newline])
+                pendingLine.removeSubrange(...newline)
+                pendingLineSearchOffset = 0
+                guard !line.isEmpty,
+                      let frame = CloudTuiManualIOFrameDecoder().decode(line) else { continue }
+                let continuation = nextFrameContinuation
+                nextFrameContinuation = nil
+                suspendReadSourceLocked()
+                continuation?.resume(returning: frame)
+                return
+            }
+            // Only newly read bytes need scanning while a large line arrives.
+            pendingLineSearchOffset = pendingLine.count
+            // Retain at most one protocol-sized unfinished line plus one read
+            // chunk. Complete frames stay in the socket until next() requests
+            // them; kernel/link backpressure bounds a stalled consumer's memory.
+            guard pendingLine.count <= Self.maximumLineBytes else {
+                closeLocked()
+                return
+            }
+            let count = Darwin.read(descriptor, &bytes, bytes.count)
+            if count > 0 {
+                pendingLine.append(bytes, count: count)
                 continue
             }
             if count == 0 {
@@ -266,6 +292,18 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         }
     }
 
+    private func resumeReadSourceLocked() {
+        guard readSourceSuspended, let readSource else { return }
+        readSourceSuspended = false
+        readSource.resume()
+    }
+
+    private func suspendReadSourceLocked() {
+        guard !readSourceSuspended, let readSource else { return }
+        readSourceSuspended = true
+        readSource.suspend()
+    }
+
     private func flushWritesLocked() {
         guard !closed, isConnected, descriptor >= 0 else { return }
         while let first = pendingWrites.first, !closed {
@@ -320,6 +358,7 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         closed = true
         isConnected = false
         pendingLine.removeAll(keepingCapacity: false)
+        pendingLineSearchOffset = 0
         pendingWrites.removeAll(keepingCapacity: false)
         pendingWriteOffset = 0
         pendingWriteBytes = 0
@@ -334,6 +373,10 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         }
         writeSource?.cancel()
         let source = readSource
+        if readSourceSuspended {
+            readSourceSuspended = false
+            source?.resume()
+        }
         readSource = nil
         self.descriptor = -1
         source?.cancel()
@@ -346,7 +389,9 @@ final class CloudTuiManualIOConnection: @unchecked Sendable {
         }
         let continuation = startContinuation
         startContinuation = nil
-        eventsContinuation.finish()
+        let frameContinuation = nextFrameContinuation
+        nextFrameContinuation = nil
+        frameContinuation?.resume(returning: nil)
         continuation?.resume(throwing: CancellationError())
     }
 

cmux.xcodeproj/project.pbxproj

@@ -752,6 +752,7 @@
 		C11322C10000000000000001 /* CloudTuiLegacySnapshotParser.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11322C20000000000000001 /* CloudTuiLegacySnapshotParser.swift */; };
 		C11324110000000000000001 /* CloudTuiManualIOCommand.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11324120000000000000001 /* CloudTuiManualIOCommand.swift */; };
 		C11322D10000000000000001 /* CloudTuiManualIOConnection.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11322D20000000000000001 /* CloudTuiManualIOConnection.swift */; };
+		A838AA211BDA47268C683501 /* CloudTuiManualIOConnectionTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = A838AA211BDA47268C683502 /* CloudTuiManualIOConnectionTests.swift */; };
 		C11324310000000000000001 /* CloudTuiManualIODescriptorLease.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11324320000000000000001 /* CloudTuiManualIODescriptorLease.swift */; };
 		C11324410000000000000001 /* CloudTuiManualIOFrame.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11324420000000000000001 /* CloudTuiManualIOFrame.swift */; };
 		C11324010000000000000001 /* CloudTuiManualIOFrameDecoder.swift in Sources */ = {isa = PBXBuildFile; fileRef = C11324020000000000000001 /* CloudTuiManualIOFrameDecoder.swift */; };
@@ -4337,6 +4338,7 @@
 		C11322C20000000000000001 /* CloudTuiLegacySnapshotParser.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiLegacySnapshotParser.swift; sourceTree = "<group>"; };
 		C11324120000000000000001 /* CloudTuiManualIOCommand.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIOCommand.swift; sourceTree = "<group>"; };
 		C11322D20000000000000001 /* CloudTuiManualIOConnection.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIOConnection.swift; sourceTree = "<group>"; };
+		A838AA211BDA47268C683502 /* CloudTuiManualIOConnectionTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIOConnectionTests.swift; sourceTree = "<group>"; };
 		C11324320000000000000001 /* CloudTuiManualIODescriptorLease.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIODescriptorLease.swift; sourceTree = "<group>"; };
 		C11324420000000000000001 /* CloudTuiManualIOFrame.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIOFrame.swift; sourceTree = "<group>"; };
 		C11324020000000000000001 /* CloudTuiManualIOFrameDecoder.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CloudTuiManualIOFrameDecoder.swift; sourceTree = "<group>"; };
@@ -10194,6 +10196,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef
 				EC905DADF2BA91C8C865D37B /* PaneFocusTestWindow.swift */,
 				E960400B0C66CA43D219BC5B /* PaneFocusFindResponderTests.swift */,
 				FCBB7EB399B58A6D076EAF1E /* CloudWorkspaceLayoutTranslatorTests.swift */,
+				A838AA211BDA47268C683502 /* CloudTuiManualIOConnectionTests.swift */,
 				64CB7AF7545F56E95A91B154 /* CloudNotificationSyncTests.swift */,
 				5C2A7143303749BE82E51F69 /* CloudProviderRefreshCoordinatorTests.swift */,
 				7EFB9C9E7B5D4F39873871A1 /* CloudVMStateSnapshotComparisonTests.swift */,
@@ -13945,6 +13948,7 @@ B8B056D80000000000000002 /* MobileHostIdentityTests.swift */ = {isa = PBXFileRef
 				B5C68F2FAFD21A103AFF429D /* CloudTreeMachineMenuTests.swift in Sources */,
 				C986A0030000000000000001 /* CloudTreeNativeDragOwnershipTests.swift in Sources */,
 				45F29E8BB64E4EE4BF204AC7 /* CloudTreeOneMachineManyWorkspacesTests.swift in Sources */,
+				A838AA211BDA47268C683501 /* CloudTuiManualIOConnectionTests.swift in Sources */,
 				7A0CE1000000000000000106 /* CloudTunnelBackendSelectorTests.swift in Sources */,
 				7A0CE1000000000000000730 /* CloudTunnelBannerTests.swift in Sources */,
 				7A0CE1000000000000000602 /* CloudTunnelBroadcastTests.swift in Sources */,