Exercise terminal reconnects across platforms
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
ci / test (push) Canceled after 0s
ci / fuzz-smoke (push) Canceled after 0s
ci / macos-client (macos-aarch64, macos-14) (push) Canceled after 0s
ci / macos-client (macos-x86_64, macos-13) (push) Canceled after 0s
ci / windows-client (push) Canceled after 0s
ci / package-release (linux-x86_64, ubuntu-latest, , , ) (push) Canceled after 0s
ci / package-release (macos-aarch64, macos-14, , , ) (push) Canceled after 0s
ci / package-release (macos-x86_64, macos-13, , , ) (push) Canceled after 0s
ci / package-release (windows-aarch64, windows-latest, aarch64, windows, aarch64-pc-windows-msvc) (push) Canceled after 0s
ci / package-release (windows-x86_64, windows-latest, , , ) (push) Canceled after 0s
ci / remote-bench (push) Canceled after 0s
ci / publish-gitea-release (push) Canceled after 0s
This commit is contained in:
@@ -50,6 +50,7 @@ struct CachedCredentialWire {
|
||||
enum ServerObservation {
|
||||
BulkSent,
|
||||
Input(Vec<u8>),
|
||||
Reconnected,
|
||||
Resize(u16, u16),
|
||||
}
|
||||
|
||||
@@ -224,12 +225,87 @@ fn terminal_output_backpressure_does_not_block_input_transport() {
|
||||
assert!(status.success(), "Dosh client exited with {status:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_reconnect_restores_snapshot_and_orders_reordered_frames() {
|
||||
const PROBE: &[u8] = b"DOSH_RECONNECT_INPUT\r";
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let home = dir.path().join("home");
|
||||
let cache = dir.path().join("credentials");
|
||||
fs::create_dir_all(home.join(".config/dosh")).unwrap();
|
||||
fs::create_dir_all(&cache).unwrap();
|
||||
|
||||
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
|
||||
socket
|
||||
.set_read_timeout(Some(Duration::from_secs(8)))
|
||||
.unwrap();
|
||||
let port = socket.local_addr().unwrap().port();
|
||||
write_client_fixture(&home, &cache, port);
|
||||
let (observation_tx, observation_rx) = mpsc::channel();
|
||||
let server = thread::spawn(move || run_reconnect_server(socket, observation_tx, PROBE));
|
||||
|
||||
let pty = NativePtySystem::default();
|
||||
let pair = pty
|
||||
.openpty(PtySize {
|
||||
rows: 24,
|
||||
cols: 80,
|
||||
pixel_width: 0,
|
||||
pixel_height: 0,
|
||||
})
|
||||
.unwrap();
|
||||
let mut reader = pair.master.try_clone_reader().unwrap();
|
||||
let mut writer = pair.master.take_writer().unwrap();
|
||||
let mut command = client_command(dir.path(), port);
|
||||
command.env("HOME", home.to_string_lossy().to_string());
|
||||
command.env("USERPROFILE", home.to_string_lossy().to_string());
|
||||
command.env("APPDATA", home.to_string_lossy().to_string());
|
||||
command.env("LOCALAPPDATA", home.to_string_lossy().to_string());
|
||||
command.env("TERM", "xterm-256color");
|
||||
let mut child = pair.slave.spawn_command(command).unwrap();
|
||||
drop(pair.slave);
|
||||
|
||||
let output = Arc::new(Mutex::new(Vec::new()));
|
||||
let reader_output = Arc::clone(&output);
|
||||
let reader_thread = thread::spawn(move || {
|
||||
let mut buf = [0u8; 4096];
|
||||
loop {
|
||||
match reader.read(&mut buf) {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(n) => reader_output.lock().unwrap().extend_from_slice(&buf[..n]),
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
wait_for_output(&output, b"DOSH_RECONNECT_READY", Duration::from_secs(3));
|
||||
wait_for_observation(&observation_rx, Duration::from_secs(4), |observation| {
|
||||
matches!(observation, ServerObservation::Reconnected)
|
||||
});
|
||||
wait_for_output(&output, b"DOSH_RECONNECT_SNAPSHOT", Duration::from_secs(3));
|
||||
|
||||
writer.write_all(PROBE).unwrap();
|
||||
writer.flush().unwrap();
|
||||
wait_for_input(&observation_rx, PROBE, Duration::from_secs(2));
|
||||
wait_for_output(
|
||||
&output,
|
||||
b"DOSH_ORDER_FIRST_DOSH_ORDER_SECOND_DOSH_RECONNECT_DONE",
|
||||
Duration::from_secs(3),
|
||||
);
|
||||
|
||||
let status = child.wait().unwrap();
|
||||
drop(writer);
|
||||
drop(pair.master);
|
||||
reader_thread.join().unwrap();
|
||||
server.join().unwrap();
|
||||
assert!(status.success(), "Dosh client exited with {status:?}");
|
||||
}
|
||||
|
||||
fn write_client_fixture(home: &Path, cache: &Path, port: u16) {
|
||||
let config = ClientConfig {
|
||||
server: "local".to_string(),
|
||||
dosh_host: Some("127.0.0.1".to_string()),
|
||||
dosh_port: port,
|
||||
cache_attach_tickets: false,
|
||||
reconnect_timeout_secs: 1,
|
||||
credential_cache: cache.to_string_lossy().to_string(),
|
||||
auth_preference: "ssh".to_string(),
|
||||
predict: false,
|
||||
@@ -479,6 +555,89 @@ fn run_backpressure_server(
|
||||
}
|
||||
}
|
||||
|
||||
fn run_reconnect_server(
|
||||
socket: UdpSocket,
|
||||
observations: mpsc::Sender<ServerObservation>,
|
||||
expected_input: &[u8],
|
||||
) {
|
||||
let mut resume_count = 0u64;
|
||||
let mut received_input = Vec::new();
|
||||
let mut buf = [0u8; 65535];
|
||||
loop {
|
||||
let (n, source) = socket.recv_from(&mut buf).unwrap();
|
||||
let packet = protocol::decode(&buf[..n]).unwrap();
|
||||
match packet.header.kind {
|
||||
PacketKind::ResumeRequest => {
|
||||
let plain =
|
||||
protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
|
||||
let request: protocol::ResumeRequest = protocol::from_body(&plain).unwrap();
|
||||
assert_eq!(request.session, SESSION);
|
||||
resume_count += 1;
|
||||
let bytes = if resume_count == 1 {
|
||||
b"DOSH_RECONNECT_READY".as_slice()
|
||||
} else {
|
||||
observations.send(ServerObservation::Reconnected).unwrap();
|
||||
b"DOSH_RECONNECT_SNAPSHOT".as_slice()
|
||||
};
|
||||
send_frame(
|
||||
&socket,
|
||||
source,
|
||||
PacketKind::ResumeOk,
|
||||
resume_count,
|
||||
10,
|
||||
bytes,
|
||||
true,
|
||||
false,
|
||||
);
|
||||
}
|
||||
PacketKind::Input => {
|
||||
let plain =
|
||||
protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
|
||||
let input: Input = protocol::from_body(&plain).unwrap();
|
||||
received_input.extend_from_slice(&input.bytes);
|
||||
observations
|
||||
.send(ServerObservation::Input(input.bytes))
|
||||
.unwrap();
|
||||
if contains(&received_input, expected_input) {
|
||||
send_frame(
|
||||
&socket,
|
||||
source,
|
||||
PacketKind::Frame,
|
||||
4,
|
||||
12,
|
||||
b"DOSH_ORDER_SECOND_",
|
||||
false,
|
||||
false,
|
||||
);
|
||||
send_frame(
|
||||
&socket,
|
||||
source,
|
||||
PacketKind::Frame,
|
||||
3,
|
||||
11,
|
||||
b"DOSH_ORDER_FIRST_",
|
||||
false,
|
||||
false,
|
||||
);
|
||||
send_frame(
|
||||
&socket,
|
||||
source,
|
||||
PacketKind::Frame,
|
||||
5,
|
||||
13,
|
||||
b"DOSH_RECONNECT_DONE",
|
||||
false,
|
||||
true,
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
PacketKind::Ping | PacketKind::Ack => {}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn send_frame(
|
||||
socket: &UdpSocket,
|
||||
|
||||
Reference in New Issue
Block a user