fix(connections): align journey progress with transport recovery
Track ET data-route progress separately from SSH bootstrap, preserve established hop colors, and reject stale route events. Distinguish direct Mosh recovery from full SSH reconnection while retaining whole-route failure coloring. Consolidate Journey waiting transitions and failure presentation, reduce redundant rendering, and add route lifecycle and platform regressions. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
1 parent
c43e5bc311
commit
4e91185396
26 files changed
+1088
-147
No files matched your search
@@ -53,6 +53,25 @@ struct ConnectionJourneyRouteStyleTests {
|
||||
expectRoute(journey, states: [.failed, .failed, .failed, .failed])
|
||||
}
|
||||
|
||||
@Test func etDataRouteAuthenticationFailureStillRecolorsTheWholeBootstrappedRoute() {
|
||||
let (journey, hosts) = fixture(.eternalTerminalSession)
|
||||
completeBootstrap(journey, hosts: hosts)
|
||||
let attempt = UUID()
|
||||
journey.receiveTransportRoute(
|
||||
.init(attemptID: attempt, revision: 1, stage: .opening, activeHostID: hosts[0].id))
|
||||
#expect(ConnectionJourneyRouteStyle(journey: journey).state(for: .endpoint, at: 3) == .pending)
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 2, stage: .unavailable))
|
||||
|
||||
journey.fail("ET data route authentication rejected", at: hosts[0].id)
|
||||
|
||||
#expect(journey.activeEndpointIndex == 1)
|
||||
#expect(journey.logs.last?.endpoint == hosts[0].name)
|
||||
expectRoute(journey, states: [.failed, .failed, .failed, .failed])
|
||||
#expect(ConnectionJourneyRouteStyle(journey: journey).activityEndpointIndex == nil)
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 3, stage: .unavailable))
|
||||
expectRoute(journey, states: [.failed, .failed, .failed, .failed])
|
||||
}
|
||||
|
||||
@Test func completedCheckpointStillCountsAsProgressBeforeItsEndpointFinishesConnecting() {
|
||||
let (journey, hosts) = fixture(.sshSession)
|
||||
journey.receive(.connected(hosts[0].id))
|
||||
@@ -148,6 +167,164 @@ struct ConnectionJourneyRouteStyleTests {
|
||||
#expect(ConnectionJourneyRouteStyle(journey: journey).activityEndpointIndex == 0)
|
||||
}
|
||||
|
||||
@Test(arguments: [2, 3])
|
||||
func etRebuildHighlightsOnlyTheEstablishedPrefixAndMovesActivityWithIt(jumpCount: Int) {
|
||||
let hosts = (0...jumpCount).map { Host(name: "Host \($0)", host: "host-\($0).invalid") }
|
||||
let journey = ConnectionJourney()
|
||||
let destination = hosts[jumpCount]
|
||||
journey.begin(host: destination, route: hosts, purpose: .eternalTerminalSession, username: "user")
|
||||
completeBootstrap(journey, hosts: hosts)
|
||||
journey.succeed()
|
||||
journey.hide()
|
||||
let attempt = UUID()
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 1, stage: .opening))
|
||||
journey.updateTransportRecovery(.recovering, message: "Recovering")
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
#expect(style.activityEndpointIndex == 0)
|
||||
#expect(style.state(for: .endpoint, at: 1) == .pending)
|
||||
|
||||
for establishedCount in 1...jumpCount {
|
||||
journey.receiveTransportRoute(
|
||||
.init(
|
||||
attemptID: attempt, revision: UInt64(establishedCount + 1), stage: .opening,
|
||||
establishedHostIDs: hosts.prefix(establishedCount).map(\.id),
|
||||
activeHostID: hosts[establishedCount].id,
|
||||
event: .authenticated(hosts[establishedCount - 1].id)))
|
||||
for index in 0...establishedCount {
|
||||
#expect(style.state(for: .endpoint, at: index) == .complete)
|
||||
if index > 0 { #expect(style.state(for: .segment, at: index) == .complete) }
|
||||
}
|
||||
#expect(style.state(for: .endpoint, at: establishedCount + 1) == .pending)
|
||||
#expect(style.state(for: .segment, at: establishedCount + 1) == .active)
|
||||
#expect(style.activityEndpointIndex == establishedCount)
|
||||
#expect(style.scrollTarget == "endpoint-\(establishedCount)")
|
||||
}
|
||||
journey.receiveTransportRoute(
|
||||
.init(attemptID: attempt, revision: 10, stage: .ready, establishedHostIDs: hosts.dropLast().map(\.id)))
|
||||
#expect(journey.phase == .connecting)
|
||||
expectRoute(journey, states: Array(repeating: .complete, count: hosts.count + 1))
|
||||
#expect(style.activityEndpointIndex == hosts.count)
|
||||
journey.succeed()
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
}
|
||||
|
||||
@Test func etRetainedRouteSurvivesSuspensionButRetiredRouteCannotReturnFromLateProgress() {
|
||||
let (journey, hosts) = fixture(.eternalTerminalSession)
|
||||
completeBootstrap(journey, hosts: hosts)
|
||||
let attempt = UUID()
|
||||
let ready = TerminalRouteSnapshot(
|
||||
attemptID: attempt, revision: 4, stage: .ready, establishedHostIDs: hosts.dropLast().map(\.id))
|
||||
journey.receiveTransportRoute(ready)
|
||||
journey.succeed()
|
||||
journey.hide()
|
||||
journey.updateTransportRecovery(.suspended, message: "Paused")
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
expectRoute(journey, states: [.complete, .complete, .complete, .complete])
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
journey.updateTransportRecovery(.recovering, message: "Recovering")
|
||||
#expect(style.activityEndpointIndex == 3)
|
||||
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 5, stage: .unavailable))
|
||||
journey.receiveTransportRoute(ready)
|
||||
expectRoute(journey, states: [.complete, .pending, .pending, .pending])
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
journey.receiveTransportRoute(.init(attemptID: UUID(), revision: 6, stage: .opening))
|
||||
#expect(style.activityEndpointIndex == 0)
|
||||
#expect(style.state(for: .segment, at: 1) == .active)
|
||||
#expect(style.state(for: .endpoint, at: 1) == .pending)
|
||||
}
|
||||
|
||||
@Test func etWaitingCheckpointAndSuspensionStopActivityWithoutGrayingTheEstablishedPrefix() {
|
||||
let (journey, hosts) = fixture(.eternalTerminalSession)
|
||||
journey.receiveTransportRoute(
|
||||
.init(
|
||||
attemptID: UUID(), revision: 1, stage: .opening,
|
||||
establishedHostIDs: [hosts[0].id], activeHostID: hosts[1].id))
|
||||
journey.waitForHostKey(at: hosts[1].id, fingerprint: "SHA256:test")
|
||||
journey.updateTransportRecovery(.recovering, message: "Recovering")
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
#expect(style.scrollTarget == "segment-2")
|
||||
#expect(style.state(for: .endpoint, at: 1) == .complete)
|
||||
#expect(style.state(for: .segment, at: 1) == .complete)
|
||||
journey.updateTransportRecovery(.suspended, message: "Paused")
|
||||
#expect(style.state(for: .segment, at: 2) == .pending)
|
||||
#expect(journey.checkpoint?.endpointIndex == 2)
|
||||
journey.updateTransportRecovery(.recovering, message: "Recovering")
|
||||
journey.resume()
|
||||
#expect(style.activityEndpointIndex == 1)
|
||||
#expect(style.state(for: .segment, at: 2) == .active)
|
||||
}
|
||||
|
||||
@Test(arguments: [false, true])
|
||||
func moshRecoveryDisplaysOnlyTheDirectPathIncludingAfterSuccess(coldRestore: Bool) {
|
||||
let (journey, hosts) = fixture(.moshSession)
|
||||
if !coldRestore {
|
||||
completeBootstrap(journey, hosts: hosts)
|
||||
journey.succeed()
|
||||
journey.hide()
|
||||
}
|
||||
let endpoints = journey.endpoints
|
||||
let bootstrapLogs = journey.logs
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
#expect(style.endpointIndices == [0, 1, 2, 3])
|
||||
journey.updateTransportRecovery(.recovering, message: "Recovering UDP")
|
||||
#expect(style.endpointIndices == [0, 3])
|
||||
#expect(style.endpointIndices.dropFirst().count == 1)
|
||||
#expect(style.activityEndpointIndex == 3)
|
||||
#expect(style.scrollTarget == "endpoint-3")
|
||||
#expect(style.state(for: .segment, at: 3) == .active)
|
||||
#expect(journey.endpoints == endpoints)
|
||||
#expect(Array(journey.logs.prefix(bootstrapLogs.count)) == bootstrapLogs)
|
||||
journey.updateTransportRecovery(.suspended, message: "Paused")
|
||||
#expect(style.endpointIndices == [0, 3])
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
#expect(style.state(for: .segment, at: 3) == .pending)
|
||||
journey.updateTransportRecovery(.failed, message: "Unavailable")
|
||||
#expect(style.endpointIndices == [0, 3])
|
||||
#expect(style.state(for: .segment, at: 3) == .failed)
|
||||
journey.updateTransportRecovery(.recovering, message: "Retry")
|
||||
journey.succeed()
|
||||
#expect(style.endpointIndices == [0, 3])
|
||||
#expect(style.state(for: .segment, at: 3) == .complete)
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
journey.hide()
|
||||
journey.begin(host: hosts[2], route: hosts, purpose: .moshSession, username: "user")
|
||||
#expect(style.endpointIndices == [0, 1, 2, 3])
|
||||
#expect(style.activityEndpointIndex == 0)
|
||||
}
|
||||
|
||||
@Test(arguments: [ConnectionJourney.TransportRecovery.recovering, .suspended, .failed])
|
||||
func moshFullReconnectRebuildsTheSSHRouteRatherThanReusingUDPRecovery(recovery: ConnectionJourney.TransportRecovery)
|
||||
{
|
||||
let (journey, hosts) = fixture(.moshSession)
|
||||
completeBootstrap(journey, hosts: hosts)
|
||||
journey.updateTransportRecovery(recovery, message: "Old Mosh session")
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
#expect(style.endpointIndices == [0, 3])
|
||||
|
||||
journey.begin(
|
||||
host: hosts[2], route: hosts, purpose: .moshSession, username: "user", mode: .reconnection)
|
||||
|
||||
#expect(journey.mode == .reconnection)
|
||||
#expect(journey.transportRecovery == nil)
|
||||
#expect(!journey.usesDirectRecoveryRoute)
|
||||
#expect(style.endpointIndices == [0, 1, 2, 3])
|
||||
expectRoute(journey, states: [.complete, .pending, .pending, .pending])
|
||||
#expect(style.activityEndpointIndex == 0)
|
||||
journey.waitForAuthentication(at: hosts[0].id)
|
||||
#expect(style.scrollTarget == "segment-1")
|
||||
#expect(style.activityEndpointIndex == nil)
|
||||
journey.resume()
|
||||
journey.receive(.connected(hosts[0].id))
|
||||
journey.receive(.connecting(hosts[1].id))
|
||||
#expect(style.state(for: .endpoint, at: 1) == .complete)
|
||||
#expect(style.state(for: .segment, at: 1) == .complete)
|
||||
#expect(style.state(for: .segment, at: 2) == .active)
|
||||
#expect(style.activityEndpointIndex == 1)
|
||||
#expect(style.endpointIndices == [0, 1, 2, 3])
|
||||
}
|
||||
|
||||
private func fixture(_ purpose: ConnectionPurpose) -> (ConnectionJourney, [Host]) {
|
||||
let hosts = [
|
||||
Host(name: "Jump One", host: "jump-one.example.invalid"),
|
||||
|
||||
@@ -1,12 +1,43 @@
|
||||
import ETSession
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
import Testing
|
||||
|
||||
@testable import TermindRemoteInfrastructure
|
||||
|
||||
@Suite
|
||||
struct EternalTerminalResumeTests {
|
||||
@Test func initialDataRouteFailureReportsRetirementAndPreservesTheOriginalError() async throws {
|
||||
let fixture = Fixture()
|
||||
let hostID = UUID()
|
||||
let request = SSHConnectionRequest(
|
||||
configuration: .init(
|
||||
hostID: hostID, host: "unused.invalid", port: 22, username: "unused", credentials: [],
|
||||
expectedHostKeys: []))
|
||||
do {
|
||||
_ = try await fixture.engine.prepare(
|
||||
.init(
|
||||
request: request, options: .init(), environment: [], shellIntegrationEnabled: false,
|
||||
checkpoint: nil, startup: .init(plan: nil), bootstrap: SSHConnection(),
|
||||
route: { progress in
|
||||
progress?(.connecting(hostID))
|
||||
throw TerminalTransportError.unreachable
|
||||
}))
|
||||
Issue.record("The original route failure must escape prepare")
|
||||
} catch TerminalTransportError.unreachable {
|
||||
}
|
||||
#expect(await eventually { fixture.events.routes.count == 3 })
|
||||
let snapshots = fixture.events.routes.sorted { $0.revision < $1.revision }
|
||||
#expect(snapshots.map(\.stage) == [.opening, .opening, .unavailable])
|
||||
#expect(snapshots.first?.activeHostID == hostID)
|
||||
#expect(snapshots[1].event == .connecting(hostID))
|
||||
#expect(snapshots.last?.establishedHostIDs == [])
|
||||
#expect(Set(snapshots.map(\.attemptID)).count == 1)
|
||||
#expect(fixture.events.connectionConfirmations == 0)
|
||||
await fixture.close()
|
||||
}
|
||||
|
||||
@Test func foregroundResumeConfirmsAConnectionThatStayedAliveWithoutANewVendorEvent() async throws {
|
||||
let fixture = Fixture()
|
||||
await fixture.start()
|
||||
@@ -94,6 +125,7 @@ private struct Fixture {
|
||||
self.client = client
|
||||
engine = EternalTerminalEngine {
|
||||
if case .connected = $0 { events.recordConnectionConfirmation() }
|
||||
if case .routeChanged(let snapshot) = $0 { events.recordRoute(snapshot) }
|
||||
await client.receiveEngineEvent($0)
|
||||
}
|
||||
}
|
||||
@@ -114,9 +146,15 @@ private nonisolated final class Events: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var recorded: [ResumableTerminalState] = []
|
||||
private var confirmations = 0
|
||||
private var routeSnapshots: [TerminalRouteSnapshot] = []
|
||||
var routes: [TerminalRouteSnapshot] { lock.withLock { routeSnapshots } }
|
||||
var values: [ResumableTerminalState] { lock.withLock { recorded } }
|
||||
var connectionConfirmations: Int { lock.withLock { confirmations } }
|
||||
|
||||
func recordRoute(_ snapshot: TerminalRouteSnapshot) {
|
||||
lock.withLock { routeSnapshots.append(snapshot) }
|
||||
}
|
||||
|
||||
func record(_ state: ResumableTerminalState) {
|
||||
lock.withLock { recorded.append(state) }
|
||||
}
|
||||
|
||||
@@ -121,7 +121,7 @@ struct LocalNetworkLifecycleTests {
|
||||
try await client.connect(
|
||||
request: .init(configuration: config), options: options,
|
||||
environment: [], shellIntegrationEnabled: false, columns: 80, rows: 24,
|
||||
route: { throw CancellationError() }, progress: nil)
|
||||
route: { _ in throw CancellationError() }, progress: nil)
|
||||
Issue.record("A cancelled physical connection must not open")
|
||||
} catch {
|
||||
#expect(error is CancellationError)
|
||||
|
||||
@@ -7,6 +7,48 @@ import Testing
|
||||
|
||||
@Suite
|
||||
struct ResumableTerminalEngineTests {
|
||||
@Test func routeFactsPassWhileRecoveryIsFrozenWithoutConfirmingProtocolReadiness() async throws {
|
||||
let events = EngineEvents()
|
||||
let engine = TestEngine(events: events, gates: ["retry"])
|
||||
let client = client(engine, events: events)
|
||||
try await connect(client)
|
||||
await events.emit(.recovering)
|
||||
let retrying = Task { try await client.retryNow() }
|
||||
await events.wait(for: "gated:retry")
|
||||
let attemptID = UUID()
|
||||
await events.emit(.routeChanged(.init(attemptID: attemptID, revision: 1, stage: .opening)))
|
||||
await events.emit(.routeChanged(.init(attemptID: attemptID, revision: 2, stage: .ready)))
|
||||
#expect(events.values.suffix(2) == ["route:1:opening", "route:2:ready"])
|
||||
#expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")
|
||||
await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
|
||||
await engine.release("retry")
|
||||
try await retrying.value
|
||||
#expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")
|
||||
#expect(events.values.filter { $0 == "prepareRetry" }.count == 1)
|
||||
#expect(events.values.filter { $0 == "retry" }.count == 1)
|
||||
await events.emit(.connected)
|
||||
#expect(events.values.last == "state:connected")
|
||||
await client.close()
|
||||
}
|
||||
|
||||
@Test func reorderedRouteCallbacksAndClosedClientCannotReviveAnOldRoute() async throws {
|
||||
let events = EngineEvents()
|
||||
let engine = TestEngine(events: events)
|
||||
let client = client(engine, events: events)
|
||||
try await connect(client)
|
||||
let old = UUID()
|
||||
let next = UUID()
|
||||
await events.emit(.routeChanged(.init(attemptID: old, revision: 3, stage: .unavailable)))
|
||||
await events.emit(.routeChanged(.init(attemptID: old, revision: 2, stage: .ready)))
|
||||
await events.emit(.routeChanged(.init(attemptID: next, revision: 4, stage: .opening)))
|
||||
await events.emit(.routeChanged(.init(attemptID: old, revision: 3, stage: .unavailable)))
|
||||
#expect(events.values.filter { $0.hasPrefix("route:") } == ["route:3:unavailable", "route:4:opening"])
|
||||
await client.close()
|
||||
let closed = events.values
|
||||
await events.emit(.routeChanged(.init(attemptID: next, revision: 5, stage: .ready)))
|
||||
#expect(events.values == closed)
|
||||
}
|
||||
|
||||
@Test func recoveryRetryUsesTheExistingEngineWithoutReplayingStartup() async throws {
|
||||
let events = EngineEvents()
|
||||
let engine = TestEngine(events: events)
|
||||
@@ -31,7 +73,7 @@ struct ResumableTerminalEngineTests {
|
||||
output: { _ in }, exit: { _ in })
|
||||
await channel.configureTransport(
|
||||
.mosh, options: options,
|
||||
route: {
|
||||
route: { _ in
|
||||
events.record("route")
|
||||
throw CancellationError()
|
||||
}, stateChanged: { _ in })
|
||||
@@ -62,7 +104,7 @@ struct ResumableTerminalEngineTests {
|
||||
output: { events.record("output:\(String(decoding: $0, as: UTF8.self))") }, exit: { _ in })
|
||||
await channel.configureTransport(
|
||||
.mosh, options: options, checkpoint: checkpoint,
|
||||
route: {
|
||||
route: { _ in
|
||||
events.record("route")
|
||||
throw CancellationError()
|
||||
}, stateChanged: { events.record("state:\($0)") })
|
||||
@@ -714,13 +756,14 @@ struct ResumableTerminalEngineTests {
|
||||
output: { events.record("output:\(String(decoding: $0, as: UTF8.self))") },
|
||||
stateChanged: { events.record("state:\($0)") }, shellIntegration: { _ in },
|
||||
registry: registry(engine, events: events), pathMonitor: nil,
|
||||
lifecycleRequestEnqueued: { events.record("lifecycle:queued") })
|
||||
lifecycleRequestEnqueued: { events.record("lifecycle:queued") },
|
||||
routeChanged: { events.record("route:\($0.revision):\($0.stage)") })
|
||||
}
|
||||
|
||||
private func connect(_ client: ResumableTerminalClient) async throws {
|
||||
try await client.connect(
|
||||
request: request, options: options, environment: [], shellIntegrationEnabled: true,
|
||||
columns: 80, rows: 24, route: { throw CancellationError() }, progress: nil,
|
||||
columns: 80, rows: 24, route: { _ in throw CancellationError() }, progress: nil,
|
||||
startupPreparation: { _ in
|
||||
Issue.record("Restoring a checkpoint must not prepare a new shell")
|
||||
return nil
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import Testing
|
||||
|
||||
@testable import TermindRemoteInfrastructure
|
||||
|
||||
@Suite
|
||||
struct TerminalRouteReporterTests {
|
||||
@Test func authenticationEstablishesOnlyTheIncomingPrefixAndMovesTheActiveTarget() throws {
|
||||
let first = UUID()
|
||||
let second = UUID()
|
||||
let destination = UUID()
|
||||
let events = RouteSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [first, second, destination], changed: events.record)
|
||||
let attempt = reporter.begin()
|
||||
#expect(events.values.last?.establishedHostIDs == [])
|
||||
#expect(events.values.last?.activeHostID == first)
|
||||
reporter.receive(.authenticating(first), attemptID: attempt)
|
||||
#expect(events.values.last?.establishedHostIDs == [])
|
||||
reporter.receive(.authenticated(first), attemptID: attempt)
|
||||
#expect(events.values.last?.establishedHostIDs == [first])
|
||||
#expect(events.values.last?.activeHostID == second)
|
||||
#expect(events.values.last?.event == .authenticated(first))
|
||||
#expect(events.values.last?.stage == .opening)
|
||||
|
||||
// The first hop's outbound channel completion must not move activity back to that hop.
|
||||
reporter.receive(.connected(first), attemptID: attempt)
|
||||
#expect(events.values.last?.activeHostID == second)
|
||||
reporter.receive(.proxyConnecting(second), attemptID: attempt)
|
||||
#expect(events.values.last?.establishedHostIDs == [first])
|
||||
reporter.receive(.authenticated(second), attemptID: attempt)
|
||||
#expect(events.values.last?.establishedHostIDs == [first, second])
|
||||
#expect(events.values.last?.activeHostID == destination)
|
||||
reporter.ready(attemptID: attempt)
|
||||
#expect(events.values.last?.stage == .ready)
|
||||
#expect(events.values.last?.establishedHostIDs == [first, second, destination])
|
||||
#expect(events.values.last?.activeHostID == nil)
|
||||
}
|
||||
|
||||
@Test func unrelatedOrOutOfOrderEventsCannotInventAnEstablishedPrefix() {
|
||||
let first = UUID()
|
||||
let second = UUID()
|
||||
let events = RouteSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [first, second], changed: events.record)
|
||||
let attempt = reporter.begin()
|
||||
reporter.receive(.authenticated(second), attemptID: attempt)
|
||||
reporter.receive(.authenticated(UUID()), attemptID: attempt)
|
||||
reporter.receive(.authenticated(first), attemptID: UUID())
|
||||
#expect(events.values.count == 1)
|
||||
reporter.receive(.proxyConnected(first), attemptID: attempt)
|
||||
#expect(events.values.last?.establishedHostIDs == [])
|
||||
}
|
||||
|
||||
@Test func retirementRejectsLateCallbacksAndRetriesKeepIncreasingRevisions() throws {
|
||||
let hop = UUID()
|
||||
let events = RouteSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [hop, UUID()], changed: events.record)
|
||||
let old = reporter.begin()
|
||||
reporter.receive(.authenticated(hop), attemptID: old)
|
||||
reporter.retire(attemptID: old)
|
||||
let retired = try #require(events.values.last)
|
||||
#expect(retired.stage == .unavailable)
|
||||
#expect(retired.establishedHostIDs.isEmpty)
|
||||
reporter.receive(.connected(hop), attemptID: old)
|
||||
reporter.ready(attemptID: old)
|
||||
#expect(events.values.last == retired)
|
||||
|
||||
let next = reporter.begin()
|
||||
#expect(next != old)
|
||||
#expect(events.values.last?.revision == retired.revision + 1)
|
||||
#expect(events.values.last?.establishedHostIDs == [])
|
||||
let opening = events.values
|
||||
reporter.receive(.authenticated(hop), attemptID: old)
|
||||
reporter.retire(attemptID: old)
|
||||
reporter.ready(attemptID: old)
|
||||
#expect(events.values == opening)
|
||||
reporter.receive(.authenticated(hop), attemptID: next)
|
||||
#expect(events.values.last?.establishedHostIDs == [hop])
|
||||
#expect(events.values.last?.attemptID == next)
|
||||
#expect(events.values.map(\.revision) == Array(1...UInt64(events.values.count)))
|
||||
}
|
||||
|
||||
@Test func closeInvalidatesTheCurrentRouteAndCannotBeReopened() throws {
|
||||
let host = UUID()
|
||||
let events = RouteSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [host], changed: events.record)
|
||||
let attempt = reporter.begin()
|
||||
reporter.ready(attemptID: attempt)
|
||||
reporter.close()
|
||||
let retired = try #require(events.values.last)
|
||||
#expect(retired.stage == .unavailable)
|
||||
#expect(retired.establishedHostIDs.isEmpty)
|
||||
let lateAttempt = reporter.begin()
|
||||
reporter.receive(.authenticated(host), attemptID: attempt)
|
||||
reporter.ready(attemptID: attempt)
|
||||
reporter.ready(attemptID: lateAttempt)
|
||||
reporter.retire(attemptID: attempt)
|
||||
reporter.close()
|
||||
#expect(events.values.last == retired)
|
||||
#expect(events.values.count == 3)
|
||||
}
|
||||
}
|
||||
|
||||
private nonisolated final class RouteSnapshots: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var snapshots: [TerminalRouteSnapshot] = []
|
||||
var values: [TerminalRouteSnapshot] { lock.withLock { snapshots } }
|
||||
func record(_ snapshot: TerminalRouteSnapshot) { lock.withLock { snapshots.append(snapshot) } }
|
||||
}
|
||||
@@ -15,7 +15,7 @@ struct TerminalTCPBridgeTests {
|
||||
defer { Darwin.close(peer) }
|
||||
configure(peer)
|
||||
let route = try await routed(pair[0])
|
||||
let bridge = try TerminalTCPBridge(open: { throw CancellationError() }, initial: route)
|
||||
let bridge = try TerminalTCPBridge(open: { _ in throw CancellationError() }, initial: route)
|
||||
let client = try connect(to: bridge.port)
|
||||
defer { Darwin.close(client) }
|
||||
let payload = Data((0..<131_072).map { UInt8($0 % 251) })
|
||||
@@ -44,10 +44,16 @@ struct TerminalTCPBridgeTests {
|
||||
configure(second[1])
|
||||
let initial = try await routed(first[0])
|
||||
let replacement = try await routed(second[0])
|
||||
let bridge = try TerminalTCPBridge(open: { replacement }, initial: initial)
|
||||
let snapshots = BridgeSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [UUID()], changed: snapshots.record)
|
||||
let initialAttempt = reporter.begin()
|
||||
reporter.ready(attemptID: initialAttempt)
|
||||
let bridge = try TerminalTCPBridge(
|
||||
open: { _ in replacement }, initial: initial, reporter: reporter, initialAttemptID: initialAttempt)
|
||||
let oldClient = try connect(to: bridge.port)
|
||||
try await write(oldClient, Data([1]))
|
||||
#expect(try await read(first[1], count: 1) == Data([1]))
|
||||
#expect(snapshots.values.map(\.stage) == [.opening, .ready])
|
||||
Darwin.close(oldClient)
|
||||
let newClient = try connect(to: bridge.port)
|
||||
defer { Darwin.close(newClient) }
|
||||
@@ -56,6 +62,79 @@ struct TerminalTCPBridgeTests {
|
||||
await bridge.close()
|
||||
#expect(fcntl(initial.socket, F_GETFD) == -1)
|
||||
#expect(fcntl(replacement.socket, F_GETFD) == -1)
|
||||
let reported = snapshots.values
|
||||
#expect(reported.map(\.stage) == [.opening, .ready, .unavailable, .opening, .ready, .unavailable])
|
||||
#expect(reported[2].attemptID == initialAttempt)
|
||||
#expect(reported[3].attemptID != initialAttempt)
|
||||
#expect(reported[3].establishedHostIDs.isEmpty)
|
||||
#expect(reported.map(\.revision) == Array(1...UInt64(reported.count)))
|
||||
}
|
||||
|
||||
@Test func closeDuringOpeningRejectsLateProgressAndReleasesTheLateSocket() async throws {
|
||||
var pair: [Int32] = [-1, -1]
|
||||
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0)
|
||||
defer { Darwin.close(pair[1]) }
|
||||
let lateRoute = try await routed(pair[0])
|
||||
let gate = OpeningGate()
|
||||
let snapshots = BridgeSnapshots()
|
||||
let hop = UUID()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [hop, UUID()], changed: snapshots.record)
|
||||
let bridge = try TerminalTCPBridge(open: { try await gate.open($0) }, reporter: reporter)
|
||||
let client = try connect(to: bridge.port)
|
||||
defer { Darwin.close(client) }
|
||||
#expect(await eventually { await gate.isOpening })
|
||||
await gate.emit(.authenticated(hop))
|
||||
#expect(await eventually { snapshots.values.last?.establishedHostIDs == [hop] })
|
||||
await bridge.close()
|
||||
let retired = snapshots.values
|
||||
#expect(retired.last?.stage == .unavailable)
|
||||
#expect(retired.last?.establishedHostIDs == [])
|
||||
await gate.emit(.connected(hop))
|
||||
await gate.finish(lateRoute)
|
||||
#expect(await eventually { fcntl(lateRoute.socket, F_GETFD) == -1 })
|
||||
#expect(snapshots.values == retired)
|
||||
}
|
||||
|
||||
@Test func blockedOpeningRemainsBlockedUntilExplicitRetry() async throws {
|
||||
var pair: [Int32] = [-1, -1]
|
||||
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0)
|
||||
defer { Darwin.close(pair[1]) }
|
||||
configure(pair[1])
|
||||
let route = try await routed(pair[0])
|
||||
let gate = OpeningGate()
|
||||
let snapshots = BridgeSnapshots()
|
||||
let reporter = TerminalRouteReporter(hostIDs: [UUID()], changed: snapshots.record)
|
||||
let bridge = try TerminalTCPBridge(open: { try await gate.open($0) }, reporter: reporter)
|
||||
let first = try connect(to: bridge.port)
|
||||
defer { Darwin.close(first) }
|
||||
#expect(await eventually { await gate.isOpening })
|
||||
await gate.fail()
|
||||
#expect(await eventually { snapshots.values.last?.stage == .unavailable })
|
||||
let blocked = try connect(to: bridge.port)
|
||||
defer { Darwin.close(blocked) }
|
||||
#expect(
|
||||
await eventually {
|
||||
var byte: UInt8 = 0
|
||||
return Darwin.recv(blocked, &byte, 1, 0) == 0
|
||||
})
|
||||
#expect(await gate.openCount == 1)
|
||||
await bridge.retry()
|
||||
let retried = try connect(to: bridge.port)
|
||||
defer { Darwin.close(retried) }
|
||||
#expect(await eventually { await gate.openCount == 2 })
|
||||
await gate.finish(route)
|
||||
try await write(retried, Data([7]))
|
||||
#expect(try await read(pair[1], count: 1) == Data([7]))
|
||||
#expect(snapshots.values.map(\.stage) == [.opening, .unavailable, .opening, .ready])
|
||||
await bridge.close()
|
||||
}
|
||||
|
||||
private func eventually(_ predicate: () async -> Bool) async -> Bool {
|
||||
for _ in 0..<200 {
|
||||
if await predicate() { return true }
|
||||
try? await Task.sleep(for: .milliseconds(10))
|
||||
}
|
||||
return await predicate()
|
||||
}
|
||||
|
||||
private nonisolated func routed(_ socket: Int32) async throws -> SSHRoutedSocket {
|
||||
@@ -135,3 +214,35 @@ struct TerminalTCPBridgeTests {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private nonisolated final class BridgeSnapshots: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var recorded: [TerminalRouteSnapshot] = []
|
||||
var values: [TerminalRouteSnapshot] { lock.withLock { recorded } }
|
||||
func record(_ snapshot: TerminalRouteSnapshot) { lock.withLock { recorded.append(snapshot) } }
|
||||
}
|
||||
|
||||
private actor OpeningGate {
|
||||
private var continuation: CheckedContinuation<SSHRoutedSocket, any Error>?
|
||||
private var progress: SSHRouteProgressHandler?
|
||||
private(set) var openCount = 0
|
||||
var isOpening: Bool { continuation != nil }
|
||||
|
||||
func open(_ progress: SSHRouteProgressHandler?) async throws -> SSHRoutedSocket {
|
||||
self.progress = progress
|
||||
openCount += 1
|
||||
return try await withCheckedThrowingContinuation { continuation = $0 }
|
||||
}
|
||||
|
||||
func emit(_ event: SSHRouteEvent) { progress?(event) }
|
||||
|
||||
func finish(_ socket: SSHRoutedSocket) {
|
||||
continuation?.resume(returning: socket)
|
||||
continuation = nil
|
||||
}
|
||||
|
||||
func fail() {
|
||||
continuation?.resume(throwing: CancellationError())
|
||||
continuation = nil
|
||||
}
|
||||
}
|
||||
@@ -82,6 +82,70 @@ struct TransportRecoveryJourneyTests {
|
||||
#expect(!journey.plainTextLog.contains("Must not apply"))
|
||||
}
|
||||
|
||||
@Test(arguments: ["authentication", "hostKey", "proxy", "identity", "passphrase"])
|
||||
func routeAndProtocolUpdatesDoNotDismissPendingInteraction(kind: String) {
|
||||
let (journey, host) = fixture(.eternalTerminalSession)
|
||||
switch kind {
|
||||
case "authentication": journey.waitForAuthentication(at: host.id)
|
||||
case "hostKey": journey.waitForHostKey(at: host.id, fingerprint: "SHA256:test")
|
||||
case "proxy": journey.waitForProxyAuthentication(at: host.id)
|
||||
case "identity": journey.waitForLocalIdentity(at: host.id)
|
||||
default: journey.waitForLocalPassphrase(at: host.id)
|
||||
}
|
||||
let checkpoint = journey.checkpoint
|
||||
let requirement = journey.localRequirement
|
||||
let endpoint = journey.activeEndpointIndex
|
||||
let attempt = UUID()
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 1, stage: .opening))
|
||||
for recovery in [ConnectionJourney.TransportRecovery.recovering, .suspended, .recovering] {
|
||||
journey.updateTransportRecovery(recovery, message: "Recovery changed")
|
||||
#expect(journey.phase == .waiting)
|
||||
#expect(journey.checkpoint == checkpoint)
|
||||
#expect(journey.localRequirement == requirement)
|
||||
#expect(journey.activeEndpointIndex == endpoint)
|
||||
}
|
||||
journey.receiveTransportRoute(.init(attemptID: attempt, revision: 2, stage: .ready))
|
||||
#expect(journey.phase == .waiting)
|
||||
#expect(journey.checkpoint == checkpoint)
|
||||
#expect(journey.localRequirement == requirement)
|
||||
journey.resume()
|
||||
#expect(journey.phase == .connecting)
|
||||
#expect(journey.checkpoint == nil)
|
||||
#expect(journey.localRequirement == nil)
|
||||
}
|
||||
|
||||
@Test func routeSnapshotsAreIsolatedAndDoNotReviveTheJourneyOrDeclareProtocolReadiness() {
|
||||
let (first, host) = fixture(.eternalTerminalSession)
|
||||
let second = ConnectionJourney()
|
||||
second.begin(host: host, route: [host], purpose: .eternalTerminalSession, username: "user")
|
||||
first.succeed()
|
||||
first.hide()
|
||||
let attempt = UUID()
|
||||
let ready = TerminalRouteSnapshot(attemptID: attempt, revision: 2, stage: .ready)
|
||||
first.receiveTransportRoute(ready)
|
||||
#expect(!first.isPresented)
|
||||
#expect(second.transportRoute == nil)
|
||||
#expect(second.phase == .connecting)
|
||||
first.updateTransportRecovery(.recovering, message: "Recovering")
|
||||
first.receiveTransportRoute(ready)
|
||||
#expect(first.phase == .connecting)
|
||||
first.receiveTransportRoute(.init(attemptID: attempt, revision: 3, stage: .unavailable))
|
||||
first.receiveTransportRoute(ready)
|
||||
#expect(first.transportRoute?.stage == .unavailable)
|
||||
first.begin(host: host, route: [host], purpose: .eternalTerminalSession, username: "user")
|
||||
#expect(first.transportRoute == nil)
|
||||
first.receiveTransportRoute(.init(attemptID: UUID(), revision: 1, stage: .opening))
|
||||
#expect(first.transportRoute?.stage == .opening)
|
||||
}
|
||||
|
||||
@Test(arguments: [ConnectionPurpose.moshSession, .sshSession])
|
||||
func nonETJourneysIgnoreDataRouteFacts(purpose: ConnectionPurpose) {
|
||||
let (journey, _) = fixture(purpose)
|
||||
journey.receiveTransportRoute(.init(attemptID: UUID(), revision: 1, stage: .ready))
|
||||
#expect(journey.transportRoute == nil)
|
||||
#expect(journey.phase == .connecting)
|
||||
}
|
||||
|
||||
private func fixture(_ purpose: ConnectionPurpose) -> (ConnectionJourney, Host) {
|
||||
let host = Host(name: "Server", host: "example.invalid")
|
||||
let journey = ConnectionJourney()
|
||||
|
||||
@@ -104,6 +104,8 @@ public final class ConnectionJourney {
|
||||
public private(set) var endpoints: [ConnectionEndpoint] = []
|
||||
public private(set) var phase: Phase = .hidden
|
||||
public private(set) var transportRecovery: TransportRecovery?
|
||||
public private(set) var transportRoute: TerminalRouteSnapshot?
|
||||
public private(set) var usesDirectRecoveryRoute = false
|
||||
public private(set) var activeEndpointIndex = 0
|
||||
public private(set) var completedEndpointIndex = 0
|
||||
public private(set) var checkpoint: Checkpoint?
|
||||
@@ -145,6 +147,8 @@ public final class ConnectionJourney {
|
||||
) {
|
||||
completionTask?.cancel()
|
||||
transportRecovery = nil
|
||||
transportRoute = nil
|
||||
usesDirectRecoveryRoute = false
|
||||
let nextEndpoints =
|
||||
[
|
||||
ConnectionEndpoint(
|
||||
@@ -193,6 +197,8 @@ public final class ConnectionJourney {
|
||||
public func beginSerial(name: String, address: String, mode: Mode = .connection) {
|
||||
completionTask?.cancel()
|
||||
transportRecovery = nil
|
||||
transportRoute = nil
|
||||
usesDirectRecoveryRoute = false
|
||||
purpose = .serialSession
|
||||
self.mode = mode
|
||||
endpoints = [
|
||||
@@ -227,6 +233,8 @@ public final class ConnectionJourney {
|
||||
) {
|
||||
completionTask?.cancel()
|
||||
transportRecovery = nil
|
||||
transportRoute = nil
|
||||
usesDirectRecoveryRoute = false
|
||||
purpose = .telnetSession
|
||||
self.mode = mode
|
||||
let address = host.displayAddress(for: .telnet)
|
||||
@@ -280,7 +288,21 @@ public final class ConnectionJourney {
|
||||
log(endpoint: endpoints[destination].name, message: message)
|
||||
}
|
||||
|
||||
public func receive(_ event: SSHRouteEvent) {
|
||||
/// Snapshots describe the data route, not the earlier SSH bootstrap or terminal readiness.
|
||||
public func receiveTransportRoute(_ snapshot: TerminalRouteSnapshot) {
|
||||
guard purpose == .eternalTerminalSession,
|
||||
transportRoute.map({ snapshot.revision > $0.revision }) ?? true
|
||||
else { return }
|
||||
if transportRoute?.attemptID != snapshot.attemptID || snapshot.stage == .unavailable {
|
||||
completedCheckpoints = [:]
|
||||
}
|
||||
transportRoute = snapshot
|
||||
if let event = snapshot.event, let (index, message) = details(for: event) {
|
||||
log(endpoint: endpoints[index].name, message: message)
|
||||
}
|
||||
}
|
||||
|
||||
private func details(for event: SSHRouteEvent) -> (Int, String)? {
|
||||
let hostID: Host.ID
|
||||
let message: String
|
||||
switch event {
|
||||
@@ -315,8 +337,11 @@ public final class ConnectionJourney {
|
||||
isProxyEvent
|
||||
? proxySourceEndpointIndex(for: hostID)
|
||||
: endpoints.firstIndex { $0.id == .host(hostID) }
|
||||
guard let index else { return }
|
||||
return index.map { ($0, message) }
|
||||
}
|
||||
|
||||
public func receive(_ event: SSHRouteEvent) {
|
||||
guard let (index, message) = details(for: event) else { return }
|
||||
switch event {
|
||||
case .proxyConnecting:
|
||||
activeEndpointIndex = max(completedEndpointIndex, index)
|
||||
@@ -356,17 +381,7 @@ public final class ConnectionJourney {
|
||||
/// Identity. This is not a server challenge, so the untouched route stays
|
||||
/// gray and no midpoint checkpoint is introduced.
|
||||
public func waitForLocalIdentity(at hostID: Host.ID, failure: String? = nil) {
|
||||
guard let index = endpoints.firstIndex(where: { $0.id == .host(hostID) }) else { return }
|
||||
activeEndpointIndex = 0
|
||||
completedEndpointIndex = 0
|
||||
checkpoint = nil
|
||||
localRequirement = .identity
|
||||
localRequirementEndpointIndex = 0
|
||||
phase = .waiting
|
||||
log(
|
||||
endpoint: endpoints[index].name,
|
||||
message: failure ?? String(localized: "Waiting for an Identity")
|
||||
)
|
||||
waitForLocalRequirement(.identity, at: hostID, message: failure ?? String(localized: "Waiting for an Identity"))
|
||||
}
|
||||
|
||||
public func waitForTailscaleUsername(at hostID: Host.ID, failure: String? = nil) {
|
||||
@@ -375,17 +390,19 @@ public final class ConnectionJourney {
|
||||
}
|
||||
|
||||
public func waitForLocalPassphrase(at hostID: Host.ID, failure: String? = nil) {
|
||||
waitForLocalRequirement(
|
||||
.passphrase, at: hostID, message: failure ?? String(localized: "Waiting for a Passphrase"))
|
||||
}
|
||||
|
||||
private func waitForLocalRequirement(_ requirement: LocalRequirement, at hostID: Host.ID, message: String) {
|
||||
guard let index = endpoints.firstIndex(where: { $0.id == .host(hostID) }) else { return }
|
||||
activeEndpointIndex = 0
|
||||
completedEndpointIndex = 0
|
||||
checkpoint = nil
|
||||
localRequirement = .passphrase
|
||||
localRequirement = requirement
|
||||
localRequirementEndpointIndex = 0
|
||||
phase = .waiting
|
||||
log(
|
||||
endpoint: endpoints[index].name,
|
||||
message: failure ?? String(localized: "Waiting for a Passphrase")
|
||||
)
|
||||
log(endpoint: endpoints[index].name, message: message)
|
||||
}
|
||||
|
||||
public func waitForHostKey(at hostID: Host.ID, fingerprint: String) {
|
||||
@@ -444,23 +461,27 @@ public final class ConnectionJourney {
|
||||
}
|
||||
}
|
||||
|
||||
/// Recovery retains the original route and log, but does not replay SSH authentication.
|
||||
/// Protocol recovery does not replay bootstrap or dismiss an in-flight route interaction.
|
||||
public func updateTransportRecovery(_ state: TransportRecovery, message: String) {
|
||||
guard purpose == .moshSession || purpose == .eternalTerminalSession else { return }
|
||||
guard transportRecovery != state || (state == .failed && phase != .failed(message)) else { return }
|
||||
completionTask?.cancel()
|
||||
let isWaiting = phase == .waiting && (checkpoint != nil || localRequirement != nil)
|
||||
if transportRecovery == nil { completedCheckpoints = [:] }
|
||||
transportRecovery = state
|
||||
usesDirectRecoveryRoute = purpose == .moshSession
|
||||
mode = .reconnection
|
||||
checkpoint = nil
|
||||
completedCheckpoints = [:]
|
||||
localRequirement = nil
|
||||
localRequirementEndpointIndex = nil
|
||||
activeEndpointIndex = max(0, endpoints.count - 1)
|
||||
if !isWaiting {
|
||||
checkpoint = nil
|
||||
localRequirement = nil
|
||||
localRequirementEndpointIndex = nil
|
||||
activeEndpointIndex = max(0, endpoints.count - 1)
|
||||
}
|
||||
if state == .failed {
|
||||
fail(message)
|
||||
transportRecovery = state
|
||||
} else {
|
||||
phase = .connecting
|
||||
if !isWaiting { phase = .connecting }
|
||||
log(endpoint: destination?.name ?? purpose.logName, message: message)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
import Foundation
|
||||
import TermindDomain
|
||||
|
||||
import struct TermindDomain.Host
|
||||
|
||||
/// Last-established data-route progress, independent of SSH bootstrap and protocol readiness.
|
||||
public nonisolated struct TerminalRouteSnapshot: Equatable, Sendable {
|
||||
public enum Stage: Equatable, Sendable {
|
||||
case opening
|
||||
case ready
|
||||
case unavailable
|
||||
}
|
||||
|
||||
public let attemptID: UUID
|
||||
public let revision: UInt64
|
||||
public let stage: Stage
|
||||
public let establishedHostIDs: [Host.ID]
|
||||
public let activeHostID: Host.ID?
|
||||
public let event: SSHRouteEvent?
|
||||
|
||||
public init(
|
||||
attemptID: UUID, revision: UInt64, stage: Stage,
|
||||
establishedHostIDs: [Host.ID] = [], activeHostID: Host.ID? = nil,
|
||||
event: SSHRouteEvent? = nil
|
||||
) {
|
||||
self.attemptID = attemptID
|
||||
self.revision = revision
|
||||
self.stage = stage
|
||||
self.establishedHostIDs = establishedHostIDs
|
||||
self.activeHostID = activeHostID
|
||||
self.event = event
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import SwiftUI
|
||||
|
||||
/// Shared failure content; each platform supplies its native actions and shortcuts.
|
||||
public struct ConnectionJourneyFailureCard<Actions: View>: View {
|
||||
let message: String
|
||||
let danger: Color
|
||||
let isCompact: Bool
|
||||
let actions: Actions
|
||||
|
||||
public init(
|
||||
message: String, danger: Color = .red, isCompact: Bool = false,
|
||||
@ViewBuilder actions: () -> Actions
|
||||
) {
|
||||
self.message = message
|
||||
self.danger = danger
|
||||
self.isCompact = isCompact
|
||||
self.actions = actions()
|
||||
}
|
||||
|
||||
public var body: some View {
|
||||
VStack(alignment: .leading, spacing: 14) {
|
||||
Label(AppStrings.Connections.connectionFailed.swiftUIKey, systemImage: "exclamationmark.triangle.fill")
|
||||
.font(.headline)
|
||||
.foregroundStyle(danger)
|
||||
Text(message)
|
||||
.foregroundStyle(.secondary)
|
||||
.textSelection(.enabled)
|
||||
actions
|
||||
}
|
||||
.padding(isCompact ? 14 : 18)
|
||||
.frame(maxWidth: isCompact ? 380 : nil)
|
||||
.background(danger.opacity(0.065), in: RoundedRectangle(cornerRadius: 14))
|
||||
.overlay {
|
||||
RoundedRectangle(cornerRadius: 14).strokeBorder(danger.opacity(0.3))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,48 @@ struct ConnectionJourneyRouteStyle {
|
||||
|
||||
let journey: ConnectionJourney
|
||||
|
||||
var endpointIndices: [Int] {
|
||||
let indices = Array(journey.endpoints.indices)
|
||||
guard journey.usesDirectRecoveryRoute, let first = indices.first, let last = indices.last, first != last else {
|
||||
return indices
|
||||
}
|
||||
return [first, last]
|
||||
}
|
||||
|
||||
private var dataRoute: TerminalRouteSnapshot? {
|
||||
guard journey.phase == .connecting || journey.phase == .waiting || journey.transportRecovery != nil else {
|
||||
return nil
|
||||
}
|
||||
return journey.transportRoute
|
||||
}
|
||||
|
||||
private var completedEndpointIndex: Int {
|
||||
guard let route = dataRoute else { return journey.completedEndpointIndex }
|
||||
switch route.stage {
|
||||
case .unavailable: return 0
|
||||
case .ready: return max(0, journey.endpoints.count - 1)
|
||||
case .opening:
|
||||
var completed = 0
|
||||
for (index, hostID) in zip(journey.endpoints.indices.dropFirst(), route.establishedHostIDs) {
|
||||
guard journey.endpoints[index].role == .jump, journey.endpoints[index].hostID == hostID else { break }
|
||||
completed = index
|
||||
}
|
||||
return completed
|
||||
}
|
||||
}
|
||||
|
||||
private var activeEndpointIndex: Int {
|
||||
if journey.checkpoint != nil || journey.localRequirement != nil { return journey.activeEndpointIndex }
|
||||
guard let route = dataRoute else { return journey.activeEndpointIndex }
|
||||
switch route.stage {
|
||||
case .unavailable: return 0
|
||||
case .ready: return max(0, journey.endpoints.count - 1)
|
||||
case .opening:
|
||||
return journey.endpoints.firstIndex { $0.hostID != nil && $0.hostID == route.activeHostID }
|
||||
?? min(completedEndpointIndex + 1, max(0, journey.endpoints.count - 1))
|
||||
}
|
||||
}
|
||||
|
||||
func state(for element: Element, at index: Int) -> State {
|
||||
if case .failed = journey.phase {
|
||||
// Include the failure itself and any later progress made by SSH bootstrap.
|
||||
@@ -31,14 +73,18 @@ struct ConnectionJourneyRouteStyle {
|
||||
let reachedIndex = max(journey.activeEndpointIndex, completedIndex)
|
||||
return index <= reachedIndex ? .failed : .pending
|
||||
}
|
||||
if let recovery = journey.transportRecovery {
|
||||
switch element {
|
||||
case .endpoint:
|
||||
if index == journey.activeEndpointIndex { return .active }
|
||||
return index == 0 ? .complete : .pending
|
||||
case .segment:
|
||||
return index == journey.activeEndpointIndex && recovery == .recovering ? .active : .pending
|
||||
if let route = dataRoute {
|
||||
if index <= completedEndpointIndex { return .complete }
|
||||
if route.stage == .opening, element == .segment, index == activeEndpointIndex,
|
||||
journey.transportRecovery != .suspended
|
||||
{
|
||||
return .active
|
||||
}
|
||||
return .pending
|
||||
}
|
||||
if let recovery = journey.transportRecovery {
|
||||
if index == journey.activeEndpointIndex, recovery == .recovering { return .active }
|
||||
return element == .endpoint && index == 0 ? .complete : .pending
|
||||
}
|
||||
if index <= journey.completedEndpointIndex { return .complete }
|
||||
if element == .segment, index == journey.activeEndpointIndex { return .active }
|
||||
@@ -46,15 +92,33 @@ struct ConnectionJourneyRouteStyle {
|
||||
}
|
||||
|
||||
var activityEndpointIndex: Int? {
|
||||
guard journey.phase == .connecting, journey.checkpoint == nil,
|
||||
guard journey.phase == .connecting, journey.checkpoint == nil, journey.localRequirement == nil,
|
||||
journey.transportRecovery != .suspended,
|
||||
let lastIndex = journey.endpoints.indices.last
|
||||
else { return nil }
|
||||
if let recovery = journey.transportRecovery {
|
||||
if let route = dataRoute {
|
||||
switch route.stage {
|
||||
case .unavailable: return nil
|
||||
case .ready: return lastIndex
|
||||
case .opening: break
|
||||
}
|
||||
} else if let recovery = journey.transportRecovery {
|
||||
return recovery == .recovering ? lastIndex : nil
|
||||
}
|
||||
let candidate =
|
||||
journey.activeEndpointIndex > journey.completedEndpointIndex
|
||||
? journey.activeEndpointIndex - 1 : journey.completedEndpointIndex
|
||||
let activeIndex = activeEndpointIndex
|
||||
let completedIndex = completedEndpointIndex
|
||||
let candidate = activeIndex > completedIndex ? activeIndex - 1 : completedIndex
|
||||
return min(max(journey.endpoints.startIndex, candidate), lastIndex)
|
||||
}
|
||||
|
||||
var scrollTarget: String? {
|
||||
guard let lastIndex = endpointIndices.last else { return nil }
|
||||
let candidate =
|
||||
journey.checkpoint?.endpointIndex
|
||||
?? journey.localRequirementEndpointIndex
|
||||
?? activityEndpointIndex
|
||||
?? activeEndpointIndex
|
||||
let index = endpointIndices.contains(candidate) ? candidate : lastIndex
|
||||
return journey.checkpoint != nil && index > 0 ? "segment-\(index)" : "endpoint-\(index)"
|
||||
}
|
||||
}
|
||||
@@ -23,13 +23,7 @@ public struct ConnectionJourneyRouteView: View {
|
||||
#endif
|
||||
}
|
||||
.scrollIndicators(.hidden)
|
||||
.onChange(of: journey.activeEndpointIndex) { _, _ in
|
||||
scrollToActivity(proxy)
|
||||
}
|
||||
.onChange(of: journey.checkpoint) { _, _ in
|
||||
scrollToActivity(proxy)
|
||||
}
|
||||
.onChange(of: journey.localRequirementEndpointIndex) { _, _ in
|
||||
.onChange(of: ConnectionJourneyRouteStyle(journey: journey).scrollTarget, initial: true) { _, _ in
|
||||
scrollToActivity(proxy)
|
||||
}
|
||||
}
|
||||
@@ -43,19 +37,21 @@ public struct ConnectionJourneyRouteView: View {
|
||||
let endpointSpacing: CGFloat = 190
|
||||
let labelWidth: CGFloat = 150
|
||||
let gapBetweenEndpointFrames = endpointSpacing - labelWidth
|
||||
let routeWidth = labelWidth + CGFloat(max(0, journey.endpoints.count - 1)) * endpointSpacing
|
||||
let canvasWidth = max(viewportWidth, routeWidth + 58)
|
||||
let style = ConnectionJourneyRouteStyle(journey: journey)
|
||||
let endpointIndices = style.endpointIndices
|
||||
let activityEndpointIndex = style.activityEndpointIndex
|
||||
let routeWidth = labelWidth + CGFloat(max(0, endpointIndices.count - 1)) * endpointSpacing
|
||||
let canvasWidth = max(viewportWidth, routeWidth + 58)
|
||||
|
||||
return ZStack(alignment: .topLeading) {
|
||||
HStack(spacing: 0) {
|
||||
ForEach(journey.endpoints.indices.dropFirst(), id: \.self) { index in
|
||||
ForEach(endpointIndices.dropFirst(), id: \.self) { index in
|
||||
JourneySegmentView(
|
||||
state: style.state(for: .segment, at: index),
|
||||
checkpoint: journey.checkpoint?.endpointIndex == index
|
||||
? journey.checkpoint?.kind : journey.completedCheckpoints[index],
|
||||
checkpointIsWaiting: journey.checkpoint?.endpointIndex == index,
|
||||
reduceMotion: reduceMotion
|
||||
reduceMotion: reduceMotion || journey.transportRecovery == .suspended
|
||||
)
|
||||
.frame(width: endpointSpacing, height: nodeLaneHeight)
|
||||
.id("segment-\(index)")
|
||||
@@ -64,12 +60,13 @@ public struct ConnectionJourneyRouteView: View {
|
||||
.padding(.leading, labelWidth / 2)
|
||||
|
||||
HStack(alignment: .top, spacing: gapBetweenEndpointFrames) {
|
||||
ForEach(Array(journey.endpoints.enumerated()), id: \.element.id) { index, endpoint in
|
||||
ForEach(endpointIndices.map { (index: $0, endpoint: journey.endpoints[$0]) }, id: \.endpoint.id) {
|
||||
index, endpoint in
|
||||
JourneyEndpointView(
|
||||
endpoint: endpoint,
|
||||
purpose: journey.purpose,
|
||||
state: style.state(for: .endpoint, at: index),
|
||||
showsActivity: style.activityEndpointIndex == index,
|
||||
showsActivity: activityEndpointIndex == index,
|
||||
localRequirement: journey.localRequirementEndpointIndex == index
|
||||
? journey.localRequirement : nil,
|
||||
reduceMotion: reduceMotion
|
||||
@@ -84,17 +81,13 @@ public struct ConnectionJourneyRouteView: View {
|
||||
}
|
||||
|
||||
private func scrollToActivity(_ proxy: ScrollViewProxy) {
|
||||
guard !reduceMotion, let lastIndex = journey.endpoints.indices.last else { return }
|
||||
let candidate =
|
||||
journey.checkpoint?.endpointIndex
|
||||
?? journey.localRequirementEndpointIndex
|
||||
?? journey.activeEndpointIndex
|
||||
let index = min(max(journey.endpoints.startIndex, candidate), lastIndex)
|
||||
let target =
|
||||
journey.checkpoint != nil && index > journey.endpoints.startIndex
|
||||
? "segment-\(index)" : "endpoint-\(index)"
|
||||
withAnimation(.easeInOut(duration: 0.24)) {
|
||||
guard let target = ConnectionJourneyRouteStyle(journey: journey).scrollTarget else { return }
|
||||
if reduceMotion {
|
||||
proxy.scrollTo(target, anchor: .center)
|
||||
} else {
|
||||
withAnimation(.easeInOut(duration: 0.24)) {
|
||||
proxy.scrollTo(target, anchor: .center)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -239,26 +239,26 @@ public struct ConnectionJourneyReconnectStatus: View {
|
||||
}
|
||||
|
||||
public var body: some View {
|
||||
TimelineView(.periodic(from: .now, by: 1)) { context in
|
||||
VStack(spacing: 12) {
|
||||
HStack(spacing: 9) {
|
||||
Image(systemName: "arrow.trianglehead.2.clockwise.rotate.90")
|
||||
.foregroundStyle(accent)
|
||||
VStack(spacing: 12) {
|
||||
HStack(spacing: 9) {
|
||||
Image(systemName: "arrow.trianglehead.2.clockwise.rotate.90")
|
||||
.foregroundStyle(accent)
|
||||
TimelineView(.periodic(from: .now, by: 1)) { context in
|
||||
Text(statusText(at: context.date)).font(.headline)
|
||||
}
|
||||
Text(state.lastError)
|
||||
.font(.callout)
|
||||
.foregroundStyle(.secondary)
|
||||
.lineLimit(2)
|
||||
.multilineTextAlignment(.center)
|
||||
.textSelection(.enabled)
|
||||
ConnectionJourneyActions {
|
||||
retryButton
|
||||
stopButton
|
||||
}
|
||||
}
|
||||
.padding(.top, 4)
|
||||
Text(state.lastError)
|
||||
.font(.callout)
|
||||
.foregroundStyle(.secondary)
|
||||
.lineLimit(2)
|
||||
.multilineTextAlignment(.center)
|
||||
.textSelection(.enabled)
|
||||
ConnectionJourneyActions {
|
||||
retryButton
|
||||
stopButton
|
||||
}
|
||||
}
|
||||
.padding(.top, 4)
|
||||
}
|
||||
|
||||
private var retryButton: some View {
|
||||
|
||||
@@ -26,6 +26,7 @@ actor EternalTerminalEngine: ResumableTerminalEngine {
|
||||
private var session: ETTerminalSession?
|
||||
private var credentials: ETCredentials?
|
||||
private var bridge: TerminalTCPBridge?
|
||||
private var routeReporter: TerminalRouteReporter?
|
||||
private var reader: Task<Void, Never>?
|
||||
private var observer: Task<Void, Never>?
|
||||
private var connectionState: ETConnectionState = .idle
|
||||
@@ -36,13 +37,36 @@ actor EternalTerminalEngine: ResumableTerminalEngine {
|
||||
}
|
||||
|
||||
func prepare(_ context: ResumableTerminalEngineContext) async throws -> ResumableTerminalEnginePreparation {
|
||||
try checkOpen()
|
||||
let payload = try context.checkpoint.map(Self.decodeCheckpoint)
|
||||
let socket = try await context.route()
|
||||
let reporter = TerminalRouteReporter(
|
||||
hostIDs: context.request.configuration.jumps.map(\.id) + [context.request.configuration.hostID]
|
||||
) { [receive] snapshot in
|
||||
Task { await receive(.routeChanged(snapshot)) }
|
||||
}
|
||||
routeReporter = reporter
|
||||
let attemptID = reporter.begin()
|
||||
let socket: SSHRoutedSocket
|
||||
do {
|
||||
socket = try await context.route { reporter.receive($0, attemptID: attemptID) }
|
||||
} catch {
|
||||
reporter.retire(attemptID: attemptID)
|
||||
throw error
|
||||
}
|
||||
guard !closed else {
|
||||
socket.close()
|
||||
throw CancellationError()
|
||||
}
|
||||
let bridge = try TerminalTCPBridge(open: context.route, initial: socket)
|
||||
reporter.ready(attemptID: attemptID)
|
||||
let bridge: TerminalTCPBridge
|
||||
do {
|
||||
bridge = try TerminalTCPBridge(
|
||||
open: context.route, initial: socket, reporter: reporter, initialAttemptID: attemptID)
|
||||
} catch {
|
||||
socket.close()
|
||||
reporter.retire(attemptID: attemptID)
|
||||
throw error
|
||||
}
|
||||
self.bridge = bridge
|
||||
if let payload {
|
||||
credentials = ETCredentials(clientID: payload.clientID, passkey: payload.passkey)
|
||||
@@ -167,6 +191,7 @@ actor EternalTerminalEngine: ResumableTerminalEngine {
|
||||
func close() async {
|
||||
guard !closed else { return }
|
||||
closed = true
|
||||
routeReporter?.close()
|
||||
reader?.cancel()
|
||||
observer?.cancel()
|
||||
await session?.close()
|
||||
|
||||
@@ -4,13 +4,15 @@ import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
public typealias RouteFactory = @Sendable () async throws -> SSHRoutedSocket
|
||||
public typealias RouteFactory = @Sendable (SSHRouteProgressHandler?) async throws -> SSHRoutedSocket
|
||||
public typealias StateHandler = @Sendable (ResumableTerminalState) -> Void
|
||||
|
||||
private let transport: HostTerminalProtocol
|
||||
private let registry: ResumableTerminalEngineRegistry
|
||||
private let output: SSHPTYChannel.OutputHandler
|
||||
private let stateChanged: StateHandler
|
||||
private let routeChanged: @Sendable (TerminalRouteSnapshot) -> Void
|
||||
private var routeRevision: UInt64?
|
||||
private let integrationChanged: SSHPTYChannel.ShellIntegrationHandler
|
||||
private let lifecycleRequestEnqueued: (@Sendable () -> Void)?
|
||||
private var engine: (any ResumableTerminalEngine)?
|
||||
@@ -58,11 +60,13 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
output: @escaping SSHPTYChannel.OutputHandler,
|
||||
stateChanged: @escaping StateHandler,
|
||||
shellIntegration: @escaping SSHPTYChannel.ShellIntegrationHandler,
|
||||
registry: ResumableTerminalEngineRegistry = .live
|
||||
registry: ResumableTerminalEngineRegistry = .live,
|
||||
routeChanged: @escaping @Sendable (TerminalRouteSnapshot) -> Void = { _ in }
|
||||
) {
|
||||
self.init(
|
||||
transport: transport, checkpoint: checkpoint, output: output, stateChanged: stateChanged,
|
||||
shellIntegration: shellIntegration, registry: registry, pathMonitor: NWPathMonitor())
|
||||
shellIntegration: shellIntegration, registry: registry, pathMonitor: NWPathMonitor(),
|
||||
routeChanged: routeChanged)
|
||||
}
|
||||
|
||||
init(
|
||||
@@ -72,13 +76,15 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
shellIntegration: @escaping SSHPTYChannel.ShellIntegrationHandler,
|
||||
registry: ResumableTerminalEngineRegistry,
|
||||
pathMonitor: NWPathMonitor?,
|
||||
lifecycleRequestEnqueued: (@Sendable () -> Void)? = nil
|
||||
lifecycleRequestEnqueued: (@Sendable () -> Void)? = nil,
|
||||
routeChanged: @escaping @Sendable (TerminalRouteSnapshot) -> Void = { _ in }
|
||||
) {
|
||||
self.transport = transport
|
||||
self.registry = registry
|
||||
self.pendingCheckpoint = checkpoint
|
||||
self.output = output
|
||||
self.stateChanged = stateChanged
|
||||
self.routeChanged = routeChanged
|
||||
self.integrationChanged = shellIntegration
|
||||
self.pathMonitor = pathMonitor
|
||||
self.lifecycleRequestEnqueued = lifecycleRequestEnqueued
|
||||
@@ -366,6 +372,10 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
guard !closed else { return }
|
||||
switch event {
|
||||
case .output(let data): await receive(data)
|
||||
case .routeChanged(let snapshot):
|
||||
guard routeRevision.map({ snapshot.revision > $0 }) ?? true else { return }
|
||||
routeRevision = snapshot.revision
|
||||
routeChanged(snapshot)
|
||||
case .activity, .connected:
|
||||
engineState = .connected
|
||||
if !frozen { publish(.connected) }
|
||||
|
||||
@@ -4,6 +4,7 @@ import TermindDomain
|
||||
|
||||
nonisolated enum ResumableTerminalEngineEvent: Sendable {
|
||||
case output(Data)
|
||||
case routeChanged(TerminalRouteSnapshot)
|
||||
case activity, connected, recovering, failed, ended
|
||||
}
|
||||
|
||||
|
||||
@@ -747,6 +747,7 @@ public actor SSHPTYChannel {
|
||||
private var transportRoute: ResumableTerminalClient.RouteFactory?
|
||||
private var transportControl: RemoteSessionCommandExecutor?
|
||||
private var transportState: ResumableTerminalClient.StateHandler = { _ in }
|
||||
private var transportRouteChanged: @Sendable (TerminalRouteSnapshot) -> Void = { _ in }
|
||||
|
||||
public init(
|
||||
output: @escaping OutputHandler, exit: @escaping ExitHandler,
|
||||
@@ -782,7 +783,8 @@ public actor SSHPTYChannel {
|
||||
guard let transportRoute else { throw TerminalTransportError.unsupportedRoute }
|
||||
let client = ResumableTerminalClient(
|
||||
transport: transportKind, checkpoint: transportCheckpoint, output: transportOutput,
|
||||
stateChanged: transportState, shellIntegration: transportIntegration, registry: transportEngines)
|
||||
stateChanged: transportState, shellIntegration: transportIntegration, registry: transportEngines,
|
||||
routeChanged: transportRouteChanged)
|
||||
resumable = client
|
||||
try await client.connect(
|
||||
request: request, options: transportOptions, environment: environmentVariables,
|
||||
@@ -819,7 +821,8 @@ public actor SSHPTYChannel {
|
||||
checkpoint: TerminalTransportCheckpoint? = nil,
|
||||
route: @escaping ResumableTerminalClient.RouteFactory,
|
||||
control: RemoteSessionCommandExecutor? = nil,
|
||||
stateChanged: @escaping ResumableTerminalClient.StateHandler
|
||||
stateChanged: @escaping ResumableTerminalClient.StateHandler,
|
||||
routeChanged: @escaping @Sendable (TerminalRouteSnapshot) -> Void = { _ in }
|
||||
) {
|
||||
transportKind = kind
|
||||
transportOptions = options
|
||||
@@ -827,6 +830,7 @@ public actor SSHPTYChannel {
|
||||
transportRoute = route
|
||||
transportControl = control
|
||||
transportState = stateChanged
|
||||
transportRouteChanged = routeChanged
|
||||
}
|
||||
|
||||
/// Share route/control wiring while the owner supplies its generation-fenced request factory.
|
||||
@@ -834,18 +838,19 @@ public actor SSHPTYChannel {
|
||||
_ kind: HostTerminalProtocol, options: HostTerminalTransports,
|
||||
checkpoint: TerminalTransportCheckpoint? = nil,
|
||||
requestFactory: @escaping @Sendable (_ authenticatingTarget: Bool) async throws -> SSHConnectionRequest,
|
||||
stateChanged: @escaping ResumableTerminalClient.StateHandler
|
||||
stateChanged: @escaping ResumableTerminalClient.StateHandler,
|
||||
routeChanged: @escaping @Sendable (TerminalRouteSnapshot) -> Void = { _ in }
|
||||
) {
|
||||
configureTransport(
|
||||
kind, options: options, checkpoint: checkpoint,
|
||||
route: {
|
||||
route: { progress in
|
||||
try await TerminalTransportRoute.open(
|
||||
using: requestFactory(false), port: options.eternalTerminalPort)
|
||||
using: requestFactory(false), port: options.eternalTerminalPort, progress: progress)
|
||||
},
|
||||
control: { command in
|
||||
try await TerminalTransportRoute.execute(command, using: requestFactory(true))
|
||||
},
|
||||
stateChanged: stateChanged)
|
||||
stateChanged: stateChanged, routeChanged: routeChanged)
|
||||
}
|
||||
|
||||
public func suspendTransport() async throws -> TerminalTransportCheckpoint {
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
import struct TermindDomain.Host
|
||||
|
||||
/// Route ownership facts, independent of the ET protocol's connection and suspension state.
|
||||
/// The bridge serializes its callbacks; the lock also covers the initial open and engine close.
|
||||
nonisolated final class TerminalRouteReporter: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private let hostIDs: [Host.ID]
|
||||
private let changed: @Sendable (TerminalRouteSnapshot) -> Void
|
||||
private var attemptID: UUID?
|
||||
private var revision: UInt64 = 0
|
||||
private var stage: TerminalRouteSnapshot.Stage = .unavailable
|
||||
private var establishedCount = 0
|
||||
private var closed = false
|
||||
|
||||
init(hostIDs: [Host.ID], changed: @escaping @Sendable (TerminalRouteSnapshot) -> Void) {
|
||||
self.hostIDs = hostIDs
|
||||
self.changed = changed
|
||||
}
|
||||
|
||||
func begin() -> UUID {
|
||||
let id = UUID()
|
||||
let snapshot = lock.withLock { () -> TerminalRouteSnapshot? in
|
||||
guard !closed else { return nil }
|
||||
attemptID = id
|
||||
stage = .opening
|
||||
establishedCount = 0
|
||||
return makeSnapshot(activeHostID: hostIDs.first)
|
||||
}
|
||||
if let snapshot { changed(snapshot) }
|
||||
return id
|
||||
}
|
||||
|
||||
func receive(_ event: SSHRouteEvent, attemptID: UUID) {
|
||||
let snapshot = lock.withLock { () -> TerminalRouteSnapshot? in
|
||||
guard !closed, self.attemptID == attemptID, stage == .opening else { return nil }
|
||||
let hostID: Host.ID
|
||||
let establishes: Bool
|
||||
switch event {
|
||||
case .authenticated(let id), .connected(let id):
|
||||
hostID = id
|
||||
establishes = true
|
||||
case .proxyConnecting(let id), .proxyConnected(let id), .connecting(let id), .authenticating(let id),
|
||||
.initializing(let id, _):
|
||||
hostID = id
|
||||
establishes = false
|
||||
}
|
||||
// Never infer an earlier hop from a later callback. Only the ordered prefix is established.
|
||||
guard hostIDs.firstIndex(of: hostID) == establishedCount else { return nil }
|
||||
if establishes { establishedCount += 1 }
|
||||
let active = establishedCount < hostIDs.count ? hostIDs[establishedCount] : nil
|
||||
return makeSnapshot(activeHostID: active, event: event)
|
||||
}
|
||||
if let snapshot { changed(snapshot) }
|
||||
}
|
||||
|
||||
func ready(attemptID: UUID) {
|
||||
let snapshot = lock.withLock { () -> TerminalRouteSnapshot? in
|
||||
guard !closed, self.attemptID == attemptID, stage == .opening else { return nil }
|
||||
stage = .ready
|
||||
establishedCount = hostIDs.count
|
||||
return makeSnapshot()
|
||||
}
|
||||
if let snapshot { changed(snapshot) }
|
||||
}
|
||||
|
||||
func retire(attemptID: UUID) {
|
||||
let snapshot = lock.withLock { () -> TerminalRouteSnapshot? in
|
||||
guard !closed, self.attemptID == attemptID, stage != .unavailable else { return nil }
|
||||
stage = .unavailable
|
||||
establishedCount = 0
|
||||
return makeSnapshot()
|
||||
}
|
||||
if let snapshot { changed(snapshot) }
|
||||
}
|
||||
|
||||
func close() {
|
||||
let snapshot = lock.withLock { () -> TerminalRouteSnapshot? in
|
||||
guard !closed else { return nil }
|
||||
closed = true
|
||||
guard attemptID != nil, stage != .unavailable else { return nil }
|
||||
stage = .unavailable
|
||||
establishedCount = 0
|
||||
return makeSnapshot()
|
||||
}
|
||||
if let snapshot { changed(snapshot) }
|
||||
}
|
||||
|
||||
private func makeSnapshot(activeHostID: Host.ID? = nil, event: SSHRouteEvent? = nil) -> TerminalRouteSnapshot {
|
||||
revision += 1
|
||||
return TerminalRouteSnapshot(
|
||||
attemptID: attemptID!, revision: revision, stage: stage,
|
||||
establishedHostIDs: Array(hostIDs.prefix(establishedCount)), activeHostID: activeHostID, event: event)
|
||||
}
|
||||
}
|
||||
@@ -5,10 +5,12 @@ import TermindApplication
|
||||
/// Adapts the ET library's TCP-only endpoint API to our routed socket owner.
|
||||
/// Socket ownership and bounded queues stay on one serial queue, including cancellation.
|
||||
nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
typealias Open = @Sendable () async throws -> SSHRoutedSocket
|
||||
typealias Open = ResumableTerminalClient.RouteFactory
|
||||
let port: UInt16
|
||||
private let queue = DispatchQueue(label: "com.akinokaede.termind.terminal-tcp-bridge")
|
||||
private let open: Open
|
||||
private let reporter: TerminalRouteReporter?
|
||||
private var routeAttemptID: UUID?
|
||||
private var listener: Int32
|
||||
private var local: Int32 = -1
|
||||
private var remote: SSHRoutedSocket?
|
||||
@@ -22,8 +24,13 @@ nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
private var closed = false
|
||||
private var openingBlocked = false
|
||||
|
||||
init(open: @escaping Open, initial: SSHRoutedSocket? = nil) throws {
|
||||
init(
|
||||
open: @escaping Open, initial: SSHRoutedSocket? = nil,
|
||||
reporter: TerminalRouteReporter? = nil, initialAttemptID: UUID? = nil
|
||||
) throws {
|
||||
self.open = open
|
||||
self.reporter = reporter
|
||||
self.routeAttemptID = initialAttemptID
|
||||
self.remote = initial
|
||||
if let initial { Self.configure(initial.socket) }
|
||||
listener = Darwin.socket(AF_INET, SOCK_STREAM, 0)
|
||||
@@ -56,6 +63,7 @@ nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
}
|
||||
|
||||
deinit {
|
||||
if let routeAttemptID { reporter?.retire(attemptID: routeAttemptID) }
|
||||
timer?.cancel()
|
||||
opening?.cancel()
|
||||
if listener >= 0 { Darwin.close(listener) }
|
||||
@@ -96,6 +104,8 @@ nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
|
||||
private func retire() {
|
||||
generation = UUID()
|
||||
if let routeAttemptID { reporter?.retire(attemptID: routeAttemptID) }
|
||||
routeAttemptID = nil
|
||||
opening?.cancel()
|
||||
opening = nil
|
||||
if local >= 0 {
|
||||
@@ -138,9 +148,18 @@ nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
|
||||
private func startOpening() {
|
||||
let token = generation
|
||||
let attemptID = reporter?.begin()
|
||||
routeAttemptID = attemptID
|
||||
let progress: SSHRouteProgressHandler = { [weak self] event in
|
||||
guard let self else { return }
|
||||
queue.async {
|
||||
guard !self.closed, self.generation == token, let attemptID else { return }
|
||||
self.reporter?.receive(event, attemptID: attemptID)
|
||||
}
|
||||
}
|
||||
opening = Task { [weak self, open] in
|
||||
do {
|
||||
let socket = try await open()
|
||||
let socket = try await open(progress)
|
||||
guard let self else {
|
||||
socket.close()
|
||||
return
|
||||
@@ -153,6 +172,7 @@ nonisolated final class TerminalTCPBridge: @unchecked Sendable {
|
||||
Self.configure(socket.socket)
|
||||
self.remote = socket
|
||||
self.opening = nil
|
||||
if let attemptID { self.reporter?.ready(attemptID: attemptID) }
|
||||
}
|
||||
} catch {
|
||||
guard let self else { return }
|
||||
|
||||
@@ -27,7 +27,9 @@ public nonisolated enum TerminalTransportRoute {
|
||||
truncated: result.stdoutTruncated || result.stderrTruncated)
|
||||
}
|
||||
|
||||
public static func open(using request: SSHConnectionRequest, port: Int) async throws -> SSHRoutedSocket {
|
||||
public static func open(
|
||||
using request: SSHConnectionRequest, port: Int, progress: SSHRouteProgressHandler? = nil
|
||||
) async throws -> SSHRoutedSocket {
|
||||
defer { request.configuration.clearSecrets() }
|
||||
try Task.checkCancellation()
|
||||
try request.configuration.accessCheck?()
|
||||
@@ -36,7 +38,7 @@ public nonisolated enum TerminalTransportRoute {
|
||||
let config = request.configuration
|
||||
return try await SSHRoutedSocket.open(
|
||||
host: config.host, port: port, proxy: config.proxy, jumps: config.jumps, timeout: 20,
|
||||
authorizeHostKey: request.authorizeHostKey, network: config.network,
|
||||
progress: progress, authorizeHostKey: request.authorizeHostKey, network: config.network,
|
||||
keyboardInteractive: request.keyboardInteractive, securityKeySigning: request.securityKeySigning,
|
||||
interactionPolicy: request.interactionPolicy, hostKeyConstraints: request.hostKeyConstraints)
|
||||
}
|
||||
|
||||
@@ -985,6 +985,10 @@ final class MobileWorkspace: Identifiable {
|
||||
purpose: connectionPurpose,
|
||||
username: usernameHint ?? hosts.identity(for: host)?.username,
|
||||
mode: reconnecting ? .reconnection : .connection)
|
||||
if resumeCheckpoint != nil, transport == .mosh {
|
||||
journey.updateTransportRecovery(
|
||||
.recovering, message: ConnectionJourney.TransportRecovery.recovering.statusMessage)
|
||||
}
|
||||
if resumeCheckpoint == nil, host.isTailscaleManaged, submissions[host.id] == nil, usernameHint == nil,
|
||||
hosts.savedTailscaleUsername(for: host) == nil
|
||||
{
|
||||
@@ -1137,6 +1141,12 @@ final class MobileWorkspace: Identifiable {
|
||||
guard let self, self.generation == token else { return }
|
||||
self.receiveTransportState(state)
|
||||
}
|
||||
},
|
||||
routeChanged: { [weak self] route in
|
||||
Task { @MainActor in
|
||||
guard let self, self.generation == token else { return }
|
||||
self.receiveTransportRoute(route)
|
||||
}
|
||||
})
|
||||
}
|
||||
await session.configureEnvironment(host.environmentVariables) { variables in
|
||||
@@ -1643,6 +1653,12 @@ final class MobileWorkspace: Identifiable {
|
||||
}
|
||||
}
|
||||
|
||||
func receiveTransportRoute(_ route: TerminalRouteSnapshot) {
|
||||
guard transportState != .ended, transportState != .closed else { return }
|
||||
prepareTransportRecoveryJourney()
|
||||
journey.receiveTransportRoute(route)
|
||||
}
|
||||
|
||||
func receiveTransportState(_ state: ResumableTerminalState, failure: String? = nil) {
|
||||
guard transportState != .ended, transportState != .closed else { return }
|
||||
guard state != transportState || failure != nil else { return }
|
||||
|
||||
@@ -160,21 +160,7 @@ private struct MobileJourneyFailureCard: View {
|
||||
@Environment(\.dynamicTypeSize) private var dynamicTypeSize
|
||||
|
||||
var body: some View {
|
||||
VStack(alignment: .leading, spacing: 14) {
|
||||
Label(AppStrings.Connections.connectionFailed.swiftUIKey, systemImage: "exclamationmark.triangle.fill")
|
||||
.font(.headline)
|
||||
.foregroundStyle(.red)
|
||||
Text(message)
|
||||
.foregroundStyle(.secondary)
|
||||
.textSelection(.enabled)
|
||||
actions
|
||||
}
|
||||
.padding(18)
|
||||
.background(Color.red.opacity(0.065), in: RoundedRectangle(cornerRadius: 14))
|
||||
.overlay {
|
||||
RoundedRectangle(cornerRadius: 14)
|
||||
.strokeBorder(Color.red.opacity(0.3))
|
||||
}
|
||||
ConnectionJourneyFailureCard(message: message) { actions }
|
||||
}
|
||||
|
||||
@ViewBuilder
|
||||
@@ -182,34 +168,34 @@ private struct MobileJourneyFailureCard: View {
|
||||
if dynamicTypeSize.isAccessibilitySize {
|
||||
VStack(spacing: 10) {
|
||||
HStack {
|
||||
Button(AppStrings.Common.cancel.swiftUIKey, action: cancel)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
Button(AppStrings.Terminal.copyLogs.swiftUIKey, action: copyLogs)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
secondaryActions
|
||||
Spacer()
|
||||
}
|
||||
Button(AppStrings.Common.retry.swiftUIKey, action: retry)
|
||||
.buttonStyle(.borderedProminent)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
.frame(maxWidth: .infinity, alignment: .trailing)
|
||||
retryButton.frame(maxWidth: .infinity, alignment: .trailing)
|
||||
}
|
||||
} else {
|
||||
HStack {
|
||||
Button(AppStrings.Common.cancel.swiftUIKey, action: cancel)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
Button(AppStrings.Terminal.copyLogs.swiftUIKey, action: copyLogs)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
secondaryActions
|
||||
Spacer()
|
||||
Button(AppStrings.Common.retry.swiftUIKey, action: retry)
|
||||
.buttonStyle(.borderedProminent)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
retryButton
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ViewBuilder
|
||||
private var secondaryActions: some View {
|
||||
Button(AppStrings.Common.cancel.swiftUIKey, action: cancel)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
Button(AppStrings.Terminal.copyLogs.swiftUIKey, action: copyLogs)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
}
|
||||
|
||||
private var retryButton: some View {
|
||||
Button(AppStrings.Common.retry.swiftUIKey, action: retry)
|
||||
.buttonStyle(.borderedProminent)
|
||||
.lineLimit(1)
|
||||
.fixedSize(horizontal: true, vertical: false)
|
||||
}
|
||||
}
|
||||
@@ -139,6 +139,62 @@ struct MobileTerminalRestorationTests {
|
||||
}
|
||||
}
|
||||
|
||||
@Test(arguments: [false, true])
|
||||
func moshConnectDistinguishesCheckpointRestoreFromFullReconnect(restoring: Bool) {
|
||||
let jump = Host(name: "Jump", host: "jump.invalid")
|
||||
var host = Host(name: "Server", host: "server.invalid")
|
||||
host.jumpHostID = jump.id
|
||||
host.terminalTransports.moshEnabled = true
|
||||
let workspace = MobileWorkspace(
|
||||
descriptor: .init(hostID: host.id, title: host.name, kind: .terminal(.mosh)),
|
||||
hosts: HostLibraryModel(hosts: [jump, host]))
|
||||
defer { workspace.disconnect() }
|
||||
workspace.journey.begin(host: host, route: [jump, host], purpose: .moshSession, username: "user")
|
||||
workspace.journey.receive(.connected(host.id))
|
||||
workspace.journey.updateTransportRecovery(.failed, message: "Previous session")
|
||||
if restoring { workspace.resumeCheckpoint = .init(transport: .mosh, payload: Data("checkpoint".utf8)) }
|
||||
|
||||
workspace.connect()
|
||||
|
||||
#expect(workspace.journey.endpoints.map(\.hostID) == [nil, jump.id, host.id])
|
||||
#expect(workspace.journey.usesDirectRecoveryRoute == restoring)
|
||||
#expect(workspace.journey.transportRecovery == (restoring ? .recovering : nil))
|
||||
#expect(workspace.journey.phase == .connecting)
|
||||
#expect(workspace.journey.completedEndpointIndex == 0)
|
||||
#expect(workspace.journey.activeEndpointIndex == (restoring ? 2 : 0))
|
||||
}
|
||||
|
||||
@Test func etRouteProgressDoesNotCompleteRecoveryAndClosedWorkspaceIgnoresLateRoutes() {
|
||||
let jump = Host(name: "Jump", host: "jump.invalid")
|
||||
let host = Host(name: "Server", host: "server.invalid")
|
||||
let workspace = MobileWorkspace(
|
||||
descriptor: .init(hostID: host.id, title: host.name, kind: .terminal(.eternalTerminal)),
|
||||
hosts: HostLibraryModel(hosts: [jump, host]))
|
||||
defer { workspace.disconnect() }
|
||||
workspace.journey.begin(
|
||||
host: host, route: [jump, host], purpose: .eternalTerminalSession, username: "user")
|
||||
let terminal = workspace.terminal
|
||||
workspace.receiveTransportState(.recovering)
|
||||
let attempt = UUID()
|
||||
workspace.receiveTransportRoute(
|
||||
.init(
|
||||
attemptID: attempt, revision: 1, stage: .opening,
|
||||
establishedHostIDs: [jump.id], activeHostID: host.id))
|
||||
#expect(workspace.journey.transportRoute?.establishedHostIDs == [jump.id])
|
||||
workspace.receiveTransportRoute(.init(attemptID: attempt, revision: 2, stage: .ready))
|
||||
#expect(workspace.journey.phase == .connecting)
|
||||
#expect(!workspace.acceptsTerminalInput)
|
||||
#expect(workspace.reconnectState == nil)
|
||||
#expect(workspace.terminal === terminal)
|
||||
workspace.receiveTransportState(.connected)
|
||||
#expect(workspace.journey.phase == .succeeded)
|
||||
workspace.disconnect()
|
||||
let retained = workspace.journey.transportRoute
|
||||
workspace.receiveTransportRoute(.init(attemptID: attempt, revision: 3, stage: .opening))
|
||||
#expect(workspace.journey.transportRoute == retained)
|
||||
#expect(!workspace.journey.isPresented)
|
||||
}
|
||||
|
||||
@Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal])
|
||||
func recoveryUsesJourneyAndRetainsTerminalWithoutSchedulingSSHReconnect(_ transport: HostTerminalProtocol)
|
||||
async throws
|
||||
|
||||
@@ -977,6 +977,12 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
|
||||
guard let self, self.connectionAttemptID == attemptID else { return }
|
||||
self.receiveTransportState(state)
|
||||
}
|
||||
},
|
||||
routeChanged: { [weak self] route in
|
||||
Task { @MainActor in
|
||||
guard let self, self.connectionAttemptID == attemptID else { return }
|
||||
self.receiveTransportRoute(route)
|
||||
}
|
||||
})
|
||||
}
|
||||
await actor.configureEnvironment(host.environmentVariables) { variables in
|
||||
@@ -1068,6 +1074,12 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable {
|
||||
}
|
||||
}
|
||||
|
||||
func receiveTransportRoute(_ route: TerminalRouteSnapshot) {
|
||||
guard transportState != .ended, transportState != .closed else { return }
|
||||
prepareTransportRecoveryJourney()
|
||||
connectionJourney.receiveTransportRoute(route)
|
||||
}
|
||||
|
||||
func receiveTransportState(_ state: ResumableTerminalState, failure: String? = nil) {
|
||||
guard transportState != .ended, transportState != .closed else { return }
|
||||
guard state != transportState || failure != nil else { return }
|
||||
|
||||
@@ -399,10 +399,7 @@ private struct JourneyFailureCard: View {
|
||||
let adaptsActions: Bool
|
||||
|
||||
var body: some View {
|
||||
VStack(alignment: .leading, spacing: 14) {
|
||||
Label(AppStrings.Connections.connectionFailed.swiftUIKey, systemImage: "exclamationmark.triangle.fill")
|
||||
.font(.headline).foregroundStyle(TermindColor.danger)
|
||||
Text(message).foregroundStyle(.secondary).textSelection(.enabled)
|
||||
ConnectionJourneyFailureCard(message: message, danger: TermindColor.danger, isCompact: compactActions) {
|
||||
if compactActions {
|
||||
HStack {
|
||||
Button(action: cancel) {
|
||||
@@ -429,13 +426,6 @@ private struct JourneyFailureCard: View {
|
||||
actions
|
||||
}
|
||||
}
|
||||
.padding(compactActions ? 14 : 18)
|
||||
.frame(maxWidth: compactActions ? 380 : nil)
|
||||
.background(TermindColor.danger.opacity(0.065), in: .rect(cornerRadius: 14))
|
||||
.overlay {
|
||||
RoundedRectangle(cornerRadius: 14)
|
||||
.strokeBorder(TermindColor.danger.opacity(0.3))
|
||||
}
|
||||
}
|
||||
|
||||
private var actions: some View {
|
||||
|
||||
@@ -7,6 +7,36 @@ import Testing
|
||||
|
||||
@Suite @MainActor
|
||||
struct TerminalTransportJourneyTests {
|
||||
@Test func etRouteProgressDoesNotCompleteRecoveryAndClosedSessionIgnoresLateRoutes() {
|
||||
let jump = Host(name: "Jump", host: "jump.invalid")
|
||||
let host = Host(name: "Server", host: "server.invalid")
|
||||
let session = SSHSessionModel(host: host)
|
||||
session.transport = .eternalTerminal
|
||||
defer { session.disconnect() }
|
||||
session.connectionJourney.begin(
|
||||
host: host, route: [jump, host], purpose: .eternalTerminalSession, username: "user")
|
||||
let surface = session.terminalState
|
||||
session.receiveTransportState(.recovering)
|
||||
let attempt = UUID()
|
||||
session.receiveTransportRoute(
|
||||
.init(
|
||||
attemptID: attempt, revision: 1, stage: .opening,
|
||||
establishedHostIDs: [jump.id], activeHostID: host.id))
|
||||
#expect(session.connectionJourney.transportRoute?.establishedHostIDs == [jump.id])
|
||||
session.receiveTransportRoute(.init(attemptID: attempt, revision: 2, stage: .ready))
|
||||
#expect(session.connectionJourney.phase == .connecting)
|
||||
#expect(!session.acceptsTerminalInput)
|
||||
#expect(session.reconnectState == nil)
|
||||
#expect(session.terminalState === surface)
|
||||
session.receiveTransportState(.connected)
|
||||
#expect(session.connectionJourney.phase == .succeeded)
|
||||
session.disconnect()
|
||||
let retained = session.connectionJourney.transportRoute
|
||||
session.receiveTransportRoute(.init(attemptID: attempt, revision: 3, stage: .opening))
|
||||
#expect(session.connectionJourney.transportRoute == retained)
|
||||
#expect(!session.connectionJourney.isPresented)
|
||||
}
|
||||
|
||||
@Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal], [false, true])
|
||||
func initialConnectedEventDoesNotDismissBootstrapJourney(_ transport: HostTerminalProtocol, hidesJourney: Bool) {
|
||||
let host = Host(name: "Server", host: "example.invalid")
|
||||
|
||||
Reference in new issue
Block a user