warp-tech/warpgate
1
1
Fork 0
mirror of https://github.com/warp-tech/warpgate.git synced 2025-09-08 23:54:23 +08:00
Code Issues Projects Releases Packages Wiki Activity

fix SCP hangups

Browse source
This commit is contained in:
Eugene 2025-08-02 20:16:04 +02:00
parent f90c529b07
commit 4d5ebe42eb
No known key found for this signature in database
GPG key ID: 5896FCBBDD1CF4F4
2 changed files with 40 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
warpgate-protocol-ssh/src/server/channel_writer.rs
View file

@ -2,23 +2,44 @@ use russh::server::Handle;
use russh::{ChannelId, CryptoVec}; use russh::{ChannelId, CryptoVec};
use tokio::sync::mpsc; use tokio::sync::mpsc;
#[derive(Debug)]
enum ChannelWriteOperation {
Data(Handle, ChannelId, CryptoVec),
ExtendedData(Handle, ChannelId, u32, CryptoVec),
}
/// Sequences data writes and runs them in background to avoid lockups /// Sequences data writes and runs them in background to avoid lockups
pub struct ChannelWriter { pub struct ChannelWriter {
tx: mpsc::UnboundedSender<(Handle, ChannelId, CryptoVec)>, tx: mpsc::UnboundedSender<ChannelWriteOperation>,
} }
impl ChannelWriter { impl ChannelWriter {
pub fn new() -> Self { pub fn new() -> Self {
let (tx, mut rx) = mpsc::unbounded_channel::<(Handle, ChannelId, CryptoVec)>(); let (tx, mut rx) = mpsc::unbounded_channel::<ChannelWriteOperation>();
tokio::spawn(async move { tokio::spawn(async move {
while let Some((handle, channel, data)) = rx.recv().await { while let Some(operation) = rx.recv().await {
let _ = handle.data(channel, data).await; match operation {
ChannelWriteOperation::Data(handle, channel, data) => {
let _ = handle.data(channel, data).await;
}
ChannelWriteOperation::ExtendedData(handle, channel, ext, data) => {
let _ = handle.extended_data(channel, ext, data).await;
}
}
} }
}); });
ChannelWriter { tx } ChannelWriter { tx }
} }
pub fn write(&self, handle: Handle, channel: ChannelId, data: CryptoVec) { pub fn write(&self, handle: Handle, channel: ChannelId, data: CryptoVec) {
let _ = self.tx.send((handle, channel, data)); let _ = self
.tx
.send(ChannelWriteOperation::Data(handle, channel, data));
}
pub fn write_extended(&self, handle: Handle, channel: ChannelId, ext: u32, data: CryptoVec) {
let _ = self.tx.send(ChannelWriteOperation::ExtendedData(
handle, channel, ext, data,
));
} }
} }

27
warpgate-protocol-ssh/src/server/session.rs
View file

@ -695,10 +695,12 @@ impl ServerSession {
} }
let server_channel_id = self.map_channel_reverse(&channel)?; let server_channel_id = self.map_channel_reverse(&channel)?;
if let Some(session) = self.session_handle.as_mut() { if let Some(session) = self.session_handle.clone() {
let _ = session self.channel_writer.write(
.data(server_channel_id.0, CryptoVec::from_slice(&data)) session,
.await; server_channel_id.0,
CryptoVec::from_slice(&data),
);
} }
} }
RCEvent::Success(channel) => { RCEvent::Success(channel) => {
@ -787,15 +789,14 @@ impl ServerSession {
} }
} }
let server_channel_id = self.map_channel_reverse(&channel)?; let server_channel_id = self.map_channel_reverse(&channel)?;
self.maybe_with_session(|handle| async move { if let Some(session) = self.session_handle.clone() {
handle self.channel_writer.write_extended(
.extended_data(server_channel_id.0, ext, CryptoVec::from_slice(&data)) session,
.await server_channel_id.0,
.map_err(|_| ()) ext,
.context("failed to send extended data")?; CryptoVec::from_slice(&data),
Ok(()) );
}) }
.await?;
} }
RCEvent::HostKeyReceived(key) => { RCEvent::HostKeyReceived(key) => {
self.emit_service_message(&format!( self.emit_service_message(&format!(