methanium/mtp
Watch
3
0
Fork
You've already forked mtp
 0
Code Issues Pull requests 3 Projects Releases 22 Packages Wiki Activity Actions

[Fix] Syncronized Webserver & Host behaviour, Fixed the 10 sec default wait on auth

Browse source
This commit is contained in:
Alex 2026-07-28 18:49:40 +02:00
commit cab2cd7a52
Signed by: alex
SSH key fingerprint: SHA256:D1+Ub8o0v4K5y1JNivW8IxEOelqLSvPmUzBbDIoZkRQ
22 changed files with 2912 additions and 1011 deletions
Download patch file Download diff file Expand all files Collapse all files

24
transport/src/connection.rs
View file

@ -57,7 +57,7 @@ impl Default for Policy {
application_close_code: 0,
open_stream_timeout: Duration::from_millis(2_000),
write_timeout: Duration::from_millis(2_000),
accept_stream_timeout: Duration::from_millis(10_000),
accept_stream_timeout: Duration::from_millis(500),
read_timeout: Duration::from_millis(30_000),
keep_alive_interval: Some(Duration::from_secs(3)),
max_idle_timeout: Some(Duration::from_secs(30)),
@ -1212,28 +1212,6 @@ mod tests {
assert_ne!(SendMode::PersistentStream, SendMode::SingleStreamPerMessage);
}
#[test]
fn test_policy_default_values() {
let p = Policy::default();
assert_eq!(p.send_mode, SendMode::PersistentStream);
assert_eq!(p.max_message_size, 16 * 1024 * 1024);
assert_eq!(p.handshake_max_message_size, 64 * 1024);
assert_eq!(p.close_frame_len, u32::MAX);
assert_eq!(p.application_close_code, 0);
assert_eq!(p.open_stream_timeout, Duration::from_millis(2_000));
assert_eq!(p.write_timeout, Duration::from_millis(2_000));
assert_eq!(p.accept_stream_timeout, Duration::from_millis(10_000));
assert_eq!(p.read_timeout, Duration::from_millis(30_000));
assert_eq!(p.keep_alive_interval, Some(Duration::from_secs(3)));
assert_eq!(p.max_idle_timeout, Some(Duration::from_secs(30)));
assert_eq!(p.force_close_delay, Duration::from_millis(300));
assert_eq!(p.persistent_stream_max_retries, 4);
assert_eq!(p.persistent_stream_retry_backoff, Duration::from_millis(20));
assert_eq!(p.receiver_queue_capacity, 1000);
assert_eq!(p.max_concurrent_stream_tasks, 128);
assert_eq!(p.max_frames_per_stream, None);
}
#[test]
fn test_policy_clone() {
let p = Policy::default();

114
transport/src/generic_connection.rs
View file

@ -166,6 +166,7 @@ pub struct GenericReceiver<C: TransportConnection> {
connection: C,
ping_sender: Arc<RwLock<Option<GenericSender<C>>>>,
max_message_size: Arc<AtomicU64>,
_accept_task: Arc<tokio::task::JoinHandle<()>>,
}
impl<C: TransportConnection> Clone for GenericReceiver<C> {
@ -177,6 +178,15 @@ impl<C: TransportConnection> Clone for GenericReceiver<C> {
connection: self.connection.clone(),
ping_sender: self.ping_sender.clone(),
max_message_size: self.max_message_size.clone(),
_accept_task: self._accept_task.clone(),
}
}
}
impl<C: TransportConnection> Drop for GenericReceiver<C> {
fn drop(&mut self) {
if Arc::strong_count(&self._accept_task) == 1 {
self._accept_task.abort();
}
}
}
@ -187,17 +197,34 @@ impl<C: TransportConnection> GenericReceiver<C> {
#[cfg(feature = "pipes")]
let (pipe_tx, pipe_rx) = mpsc::channel(policy.receiver_queue_capacity);
let ping_sender: Arc<RwLock<Option<GenericSender<C>>>> = Arc::new(RwLock::new(None));
let max_message_size = Arc::new(AtomicU64::new(policy.handshake_max_message_size));
let max_message_size = Arc::new(AtomicU64::new(
policy.handshake_max_message_size.min(policy.max_message_size),
));
let task_ping_sender = ping_sender.clone();
let task_connection = connection.clone();
let task_policy = policy.clone();
let task_max_message_size = max_message_size.clone();
tokio::spawn(async move {
let task_accept_task_tx = tx.clone();
#[cfg(feature = "pipes")]
let task_accept_task_pipe_tx = pipe_tx.clone();
let accept_task = tokio::spawn(async move {
let limit = Arc::new(Semaphore::new(
task_policy.max_concurrent_stream_tasks.max(1),
));
loop {
let stream = match timeout(
// Backpressure: stop accepting new streams if the output queue is full.
#[cfg(feature = "pipes")]
let cap_full = task_accept_task_tx.capacity() == 0
|| task_accept_task_pipe_tx.capacity() == 0;
#[cfg(not(feature = "pipes"))]
let cap_full = task_accept_task_tx.capacity() == 0;
if cap_full {
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
continue;
}
let stream = match tokio::time::timeout(
task_policy.accept_stream_timeout,
task_connection.accept_uni(),
)
@ -205,29 +232,34 @@ impl<C: TransportConnection> GenericReceiver<C> {
{
Ok(Ok(stream)) => stream,
Ok(Err(error)) => {
let _ = tx.send(Err(error)).await;
let _ = task_accept_task_tx.send(Err(error)).await;
break;
}
Err(_) => {
if task_connection.close_reason().is_some() {
let _ = tx.send(Err(CommunicationError::StreamClosed)).await;
let _ = task_accept_task_tx
.send(Err(CommunicationError::StreamClosed))
.await;
break;
} else {
continue;
}
}
};
let tx = tx.clone();
// Acquire semaphore permit BEFORE spawning the task.
let permit = match limit.clone().acquire_owned().await {
Ok(permit) => permit,
Err(_) => break,
};
let tx = task_accept_task_tx.clone();
#[cfg(feature = "pipes")]
let pipe_tx = pipe_tx.clone();
let pipe_tx = task_accept_task_pipe_tx.clone();
let policy = task_policy.clone();
let max_message_size = task_max_message_size.clone();
let permit = limit.clone();
let ping_sender = task_ping_sender.clone();
let connection = task_connection.clone();
tokio::spawn(async move {
let Ok(_permit) = permit.acquire_owned().await else {
return;
};
let _permit = permit;
let mut stream = stream;
let mut frames = 0usize;
loop {
@ -235,24 +267,36 @@ impl<C: TransportConnection> GenericReceiver<C> {
.max_frames_per_stream
.is_some_and(|max| frames >= max)
{
let close_error = CommunicationError::StreamError;
let _ = tx.send(Err(close_error.clone())).await;
connection.close(
policy.application_close_code,
b"max frames exceeded",
);
break;
}
let mut len = [0; 4];
match timeout(policy.read_timeout, stream.read_exact(&mut len)).await {
match tokio::time::timeout(policy.read_timeout, stream.read_exact(&mut len))
.await
{
Ok(Ok(())) => {}
Ok(Err(CommunicationError::StreamClosed)) => break,
Ok(Err(error)) => {
tracing::error!(
"[mtp-transport] frame header read failed: {error}"
);
tracing::warn!(%error, "MTP receive stream failed while reading frame header");
let _ = tx.send(Err(error)).await;
connection.close(
policy.application_close_code,
b"frame header read error",
);
break;
}
Err(error) => {
tracing::error!(
"[mtp-transport] frame header read timed out: {error}"
Err(_) => {
tracing::warn!("MTP receive stream timed out while reading frame header");
let _ = tx.send(Err(CommunicationError::StreamError)).await;
connection.close(
policy.application_close_code,
b"frame header timeout",
);
tracing::warn!(%error, "MTP receive stream timed out while reading frame header");
break;
}
}
@ -260,8 +304,14 @@ impl<C: TransportConnection> GenericReceiver<C> {
if len == policy.close_frame_len {
break;
}
if len as u64 > max_message_size.load(Ordering::Relaxed) {
let frame_limit = max_message_size.load(Ordering::Relaxed);
if len as u64 > frame_limit {
tracing::warn!(len, "MTP receive stream frame is too large");
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
connection.close(
policy.application_close_code,
b"frame too large",
);
break;
}
let target_len = len as usize;
@ -271,12 +321,17 @@ impl<C: TransportConnection> GenericReceiver<C> {
target_len,
"MTP receive stream could not reserve frame body"
);
let _ = tx.send(Err(CommunicationError::MessageTooLarge)).await;
connection.close(
policy.application_close_code,
b"frame allocation failed",
);
break;
}
while body.len() < target_len {
let chunk_len = (target_len - body.len()).min(16 * 1024);
let mut chunk = [0u8; 16 * 1024];
let body_read = timeout(
let body_read = tokio::time::timeout(
policy.read_timeout,
stream.read_exact(&mut chunk[..chunk_len]),
)
@ -284,16 +339,16 @@ impl<C: TransportConnection> GenericReceiver<C> {
if !matches!(&body_read, Ok(Ok(())))
|| body.try_reserve(chunk_len).is_err()
{
tracing::error!(
"[mtp-transport] frame body read failed ({} bytes): {:?}",
chunk_len,
body_read
);
tracing::warn!(
pipe_chunk_len = chunk_len,
?body_read,
"MTP receive stream failed while reading frame body"
);
let _ = tx.send(Err(CommunicationError::StreamError)).await;
connection.close(
policy.application_close_code,
b"frame body read error",
);
break;
}
body.extend_from_slice(&chunk[..chunk_len]);
@ -306,6 +361,12 @@ impl<C: TransportConnection> GenericReceiver<C> {
Ok(message) => message,
Err(_) => {
tracing::warn!("MTP receive stream contained an invalid frame");
let _ = tx.send(Err(CommunicationError::ParseCommunicationValue))
.await;
connection.close(
policy.application_close_code,
b"invalid frame",
);
break;
}
};
@ -367,6 +428,7 @@ impl<C: TransportConnection> GenericReceiver<C> {
connection,
ping_sender,
max_message_size,
_accept_task: Arc::new(accept_task),
}
}
pub async fn respond_to_pings(&self, sender: GenericSender<C>) {

51
transport/src/host.rs
View file

@ -126,31 +126,34 @@ pub async fn host_with_config(
let incoming_session = endpoint.accept().await;
tracing::debug!(elapsed = ?accept_started.elapsed(), "host accept loop: received QUIC connection");
let session_started = Instant::now();
let request = match incoming_session.await {
Ok(req) => req,
Err(e) => {
debug!("incoming WebTransport session failed: {e}");
continue;
}
};
tracing::debug!(elapsed = ?session_started.elapsed(), "host accept loop: complete WebTransport handshake");
let request_accept_started = Instant::now();
let connection = match request
.accept_with_headers([("sec-webtransport-http3-draft02", "1")])
.await
{
Ok(conn) => conn,
Err(e) => {
debug!("WebTransport request accept failed: {e}");
continue;
}
};
tracing::debug!(elapsed = ?request_accept_started.elapsed(), "host accept loop: accept WebTransport request");
let incoming_tx = incoming_tx.clone();
tokio::spawn(handle_connection(connection, incoming_tx, policy.clone()));
let policy = Arc::clone(&policy);
tokio::spawn(async move {
let session_started = Instant::now();
let request = match incoming_session.await {
Ok(req) => req,
Err(e) => {
debug!("incoming WebTransport session failed: {e}");
return;
}
};
tracing::debug!(elapsed = ?session_started.elapsed(), "host accept loop: complete WebTransport handshake");
let request_accept_started = Instant::now();
let connection = match request
.accept_with_headers([("sec-webtransport-http3-draft02", "1")])
.await
{
Ok(conn) => conn,
Err(e) => {
debug!("WebTransport request accept failed: {e}");
return;
}
};
tracing::debug!(elapsed = ?request_accept_started.elapsed(), "host accept loop: accept WebTransport request");
handle_connection(connection, incoming_tx, policy).await;
});
}
});