okooooooo/Termind
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(terminal): handle resumable transport session endings

Browse Source
This commit is contained in:
Kaede Akino committed 2026-10-16 12:27:44 +08:00
1 parent f6b2c4f444
commit 7725e7c1ae
6 files changed
+228 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

FrameworkTests/ResumableTerminalExitTests.swift
+107
View File
@@ -0,0 +1,107 @@
import ETSession
import Foundation
import MoshCore
import TermindApplication
import TermindDomain
import Testing
@testable import TermindRemoteInfrastructure
@Suite
struct ResumableTerminalExitTests {
@Test func moshStoppedDrainsFinalBytesAndEndsOnlyOnce() async throws {
let events = Events()
let client = events.client(.mosh)
let tail = Data("logout\r\n".utf8)
#expect(await !client.receiveMoshPoll(state: .stopped, operations: [.hostBytes(tail)]))
#expect(events.values == [.state(.connected), .output(tail), .state(.ended)])
#expect(await !client.receiveMoshPoll(state: .stopped, operations: [.hostBytes(tail)]))
await client.close()
#expect(events.values == [.state(.connected), .output(tail), .state(.ended)])
await #expect(throws: CancellationError.self) { try await client.write(Data("pwd\n".utf8)) }
await #expect(throws: CancellationError.self) { try await client.suspend() }
}
@Test(arguments: [MoshSessionState.running, .suspended, .failed(.timeout(timeoutMs: 1000))])
func moshSilenceSuspensionAndFailureAreNotRemoteExit(_ state: MoshSessionState) async {
let events = Events()
let client = events.client(.mosh)
_ = await client.receiveMoshPoll(state: state, operations: [])
#expect(!events.values.contains(.state(.ended)))
await client.close()
_ = await client.receiveMoshPoll(state: .stopped, operations: [])
#expect(!events.values.contains(.state(.ended)))
}
@Test func eternalEndWaitsForTheOutputConsumerBeforeClosing() async {
let events = Events()
let client = events.client(.eternalTerminal)
let output = AsyncStream<Data>.makeStream()
let states = AsyncStream<ETConnectionState>.makeStream()
await client.startEternalReaders(output: output.stream, states: states.stream)
states.continuation.yield(.connected)
#expect(await eventually { events.values.contains(.state(.connected)) })
states.continuation.yield(.sessionEnded)
let tail = Data("last terminal output\r\n".utf8)
output.continuation.yield(tail)
#expect(await eventually { events.values.contains(.output(tail)) })
#expect(!events.values.contains(.state(.ended)))
output.continuation.finish()
#expect(await eventually { events.values.contains(.state(.ended)) })
#expect(events.values == [.state(.connected), .output(tail), .state(.ended)])
states.continuation.yield(.connected)
states.continuation.yield(.sessionEnded)
states.continuation.finish()
await client.close()
#expect(events.values == [.state(.connected), .output(tail), .state(.ended)])
}
@Test func eternalDisconnectAndOutputEOFAloneDoNotCloseTheSession() async {
let events = Events()
let client = events.client(.eternalTerminal)
let output = AsyncStream<Data>.makeStream()
let states = AsyncStream<ETConnectionState>.makeStream()
await client.startEternalReaders(output: output.stream, states: states.stream)
states.continuation.yield(.disconnected)
states.continuation.yield(.reconnecting)
output.continuation.finish()
#expect(await eventually { events.values.contains(.state(.recovering)) })
states.continuation.yield(.connected)
#expect(await eventually { events.values.contains(.state(.connected)) })
#expect(!events.values.contains(.state(.ended)))
await client.close()
states.continuation.yield(.sessionEnded)
states.continuation.finish()
#expect(!events.values.contains(.state(.ended)))
}
private func eventually(_ condition: () -> Bool) async -> Bool {
for _ in 0..<200 {
if condition() { return true }
try? await Task.sleep(for: .milliseconds(10))
}
return condition()
}
}
private nonisolated final class Events: @unchecked Sendable {
enum Event: Equatable {
case output(Data)
case state(ResumableTerminalState)
}
private let lock = NSLock()
private var recorded: [Event] = []
var values: [Event] { lock.withLock { recorded } }
func client(_ transport: HostTerminalProtocol) -> ResumableTerminalClient {
ResumableTerminalClient(
transport: transport,
output: { [self] data in lock.withLock { recorded.append(.output(data)) } },
stateChanged: { [self] state in lock.withLock { recorded.append(.state(state)) } },
shellIntegration: { _ in })
}
}
Frameworks/TermindRemoteInfrastructure/SSH/ResumableTerminalClient.swift
+51 -12
View File
@@ -159,7 +159,14 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
deliver(operations)
break
}
if case .failed = await mosh.state { throw TerminalTransportError.sessionFailed }
let state = await mosh.state
if case .failed = state { throw TerminalTransportError.sessionFailed }
if state == .stopped {
rollbackMoshPID = nil
_ = await receiveMoshPoll(state: state, operations: mosh.drainHostOps())
await bootstrap.disconnect()
return
}
guard ContinuousClock.now < deadline else { throw TerminalTransportError.unreachable }
try await Task.sleep(for: .milliseconds(20))
}
@@ -181,9 +188,14 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
}
await bootstrap.disconnect()
pendingCheckpoint = nil
guard !closed else { return }
startMaintenance()
startNetworkMonitoring()
} catch {
if state == .ended {
await bootstrap.disconnect()
return
}
if let pid = rollbackMoshPID, pid > 1 {
// Only undo the server this attempt just launched; never kill a user's other sessions.
await Task.detached {
@@ -315,7 +327,7 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
bridge = nil
credentials = nil
pendingCheckpoint = nil
publish(.closed)
if state != .ended { publish(.closed) }
}
private func checkOpen() throws {
@@ -333,25 +345,45 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
}
private func publish(_ next: ResumableTerminalState) {
guard !closed || next == .closed, state != next else { return }
guard !closed || next == .closed, state != .ended, state != next else { return }
state = next
stateChanged(next)
}
private func finishRemoteExit() async {
guard !closed, state != .ended else { return }
publish(.ended)
await close()
}
private func startMoshReader(_ session: MoshClientSession) {
reader = Task { [weak self] in
while !Task.isCancelled {
guard let self else { return }
if case .failed = await session.state {
await self.publish(.failed)
return
}
await self.deliver(session.drainHostOps())
// Read terminal state before draining, so the final poll cannot miss trailing output.
let state = await session.state
guard await self.receiveMoshPoll(state: state, operations: session.drainHostOps()) else { return }
try? await Task.sleep(for: .milliseconds(16))
}
}
}
func receiveMoshPoll(state: MoshSessionState, operations: [MoshHostOp]) async -> Bool {
guard !closed else { return false }
deliver(operations)
switch state {
case .stopped:
// Remote shutdown reaches this state only with swift-mosh's shutdown-marker fix (PR #2).
await finishRemoteExit()
return false
case .failed:
publish(.failed)
return false
case .idle, .starting, .running, .suspending, .suspended, .stopping:
return true
}
}
private func deliver(_ operations: [MoshHostOp]) {
guard !closed else { return }
if !operations.isEmpty, !frozen { publish(.connected) }
@@ -361,21 +393,25 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
}
private func startEternalReaders(_ session: ETTerminalSession) {
startEternalReaders(output: session.output, states: session.stateChanges)
}
func startEternalReaders(output: AsyncStream<Data>, states: AsyncStream<ETConnectionState>) {
reader = Task { [weak self] in
for await data in session.output {
for await data in output {
guard let self, !Task.isCancelled else { return }
await self.receive(data)
}
}
observer = Task { [weak self] in
for await state in session.stateChanges {
for await state in states {
guard let self, !Task.isCancelled else { return }
await self.receiveEternalState(state)
}
}
}
private func receiveEternalState(_ state: ETConnectionState) {
private func receiveEternalState(_ state: ETConnectionState) async {
guard !closed else { return }
switch state {
case .connected:
@@ -383,7 +419,10 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
case .reconnecting, .disconnected:
if !frozen { publish(.recovering) }
case .failed: publish(.failed)
case .sessionEnded: publish(.ended)
case .sessionEnded:
// The vendor finishes output before emitting this state, but our consumer may still be draining it.
await reader?.value
await finishRemoteExit()
case .closed, .idle, .bootstrapping, .connecting: break
}
}
TFI/Features/Workspaces/MobileWorkspace.swift
+15 -3
View File
@@ -867,6 +867,7 @@ final class MobileWorkspace: Identifiable {
let reconnecting = connection.isRecovering
let token = UUID()
generation = token
transportState = nil
commandMonitor.beginConnection()
if descriptor.kind.isSSHTerminal { commandMonitor.requireScopedEvents() }
activeHistoryID = nil
@@ -1178,7 +1179,7 @@ final class MobileWorkspace: Identifiable {
pendingInitialSFTPPath = nil
journey.receive(.connected(host.id))
}
guard generation == token, !Task.isCancelled else { return }
guard generation == token, !Task.isCancelled, transportState != .ended else { return }
let additionalSnapshot = hosts.additionalConnectionSnapshot(from: config)
transferSnapshot = additionalSnapshot
if let sftp {
@@ -1543,7 +1544,8 @@ final class MobileWorkspace: Identifiable {
}
}
private func receiveTransportState(_ state: ResumableTerminalState) {
func receiveTransportState(_ state: ResumableTerminalState) {
guard transportState != .ended else { return }
transportState = state
switch state {
case .connected:
@@ -1554,7 +1556,17 @@ final class MobileWorkspace: Identifiable {
case .recovering, .suspended:
connected = false
status = String(localized: AppStrings.Connections.reconnecting)
case .failed, .ended:
case .ended:
connected = false
busy = false
connectionResumeIntent = .none
recoveryLease = nil
transportSuspension?.cancel()
transportSuspension = nil
let recoveryID = recoveryID
Task { [terminalRecoveryStore] in try? await terminalRecoveryStore.discard(recoveryID) }
handleRemoteExit(.success(0))
case .failed:
connected = false
busy = false
connection.resetRecovery()
TFITests/MobileTerminalRestorationTests.swift
+22
View File
@@ -8,6 +8,28 @@ import Testing
@Suite @MainActor
struct MobileTerminalRestorationTests {
@Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal])
func remoteEndClosesWorkspaceOnceButRecoveryAndFailureDoNot(_ transport: HostTerminalProtocol) {
let host = Host(name: "Server", host: "example.invalid")
let workspace = MobileWorkspace(
descriptor: .init(hostID: host.id, title: host.name, kind: .terminal(transport)),
hosts: HostLibraryModel(hosts: [host]))
defer { workspace.disconnect() }
var closed: [UUID] = []
workspace.didRequestClose = { closed.append($0) }
workspace.receiveTransportState(.recovering)
workspace.receiveTransportState(.failed)
#expect(closed.isEmpty)
workspace.receiveTransportState(.ended)
#expect(closed == [workspace.id])
workspace.receiveTransportState(.connected)
workspace.receiveTransportState(.ended)
#expect(closed == [workspace.id])
#expect(!workspace.connected)
#expect(!workspace.busy)
}
@Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal])
func savedTmuxReferenceSurvivesTransportRecoveryAndMissingCheckpoint(_ transport: HostTerminalProtocol) async throws
{
TFM/Features/Terminal/SSHSessionModel.swift
+8 -2
View File
@@ -449,6 +449,7 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
statusStore = store
let attemptID = UUID()
connectionAttemptID = attemptID
transportState = nil
if let ssh { Task { await ssh.disconnect() } }
ssh = nil
guard let host = connectionHost(hostID, using: store) else {
@@ -960,6 +961,7 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
)
try Task.checkCancellation()
guard connectionAttemptID == attemptID else { throw CancellationError() }
guard transportState != .ended else { return }
transferSnapshot =
statusStore?.additionalConnectionSnapshot(from: configuration)
?? SSHAdditionalConnectionSnapshot(configuration: configuration)
@@ -996,7 +998,8 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
}
}
private func receiveTransportState(_ state: ResumableTerminalState) {
func receiveTransportState(_ state: ResumableTerminalState) {
guard transportState != .ended else { return }
transportState = state
switch state {
case .connected:
@@ -1004,7 +1007,10 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
connectionJourney.hide()
case .recovering:
status = .reconnecting
case .failed, .ended:
case .ended:
// These protocols report session end, not the remote process's numeric exit status.
handleRemoteExit(.success(0))
case .failed:
connection.resetRecovery()
status = .failed(TerminalTransportError.sessionFailed.localizedDescription)
case .suspended: status = .reconnecting
TFMTests/WindowStateTests.swift
+25
View File
@@ -1520,6 +1520,31 @@ struct WindowStateTests {
#expect(app.closedWorkspaceRecords.isEmpty)
}
@Test
func resumableRemoteExitClosesOnlyItsSplitAndDoesNotTreatRecoveryAsExit() throws {
for transport in [HostTerminalProtocol.mosh, .eternalTerminal] {
let host = Host(name: "Server", host: "server.example")
let pane = Pane.sshSession(host: host, transport: transport)
let session = try #require(pane.sshSession)
let workspace = Workspace(pane: pane)
workspace.splitFocusedPane(axis: .horizontal, with: .sshSession(host: host))
let neighbor = workspace.focusedPane
let window = WindowModel(workspaces: [workspace])
session.receiveTransportState(.recovering)
session.receiveTransportState(.failed)
#expect(workspace.panes.count == 2)
session.receiveTransportState(.ended)
#expect(window.workspaces.count == 1)
#expect(workspace.panes.map(\.id) == [neighbor.id])
session.receiveTransportState(.connected)
session.receiveTransportState(.ended)
#expect(session.status == .disconnected)
#expect(workspace.panes.map(\.id) == [neighbor.id])
}
}
@Test
func normalRemoteExitClosesLastPaneAndSelectsNeighboringWorkspace() {
let host = Host(name: "Host", host: "host.example")