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

feat(wasm, native, h3): make wasm, native and h3 use unified interface
Some checks failed
CI / checks (push) Failing after 2s
Details

Browse source
This commit is contained in:
Alois 2026-08-27 15:31:55 +02:00
commit e83cd132a2
Signed by: alois
SSH key fingerprint: SHA256:GBzT2DXvAuGV9XIV5W3WrzVpjU54FThmxHXdbz95J24
13 changed files with 738 additions and 399 deletions
Download patch file Download diff file Expand all files Collapse all files

122
transport/src/connection.rs
View file

@ -4,6 +4,10 @@ use crate::framing::RetryClassifier;
use crate::pipe::PipeReader;
use mtp_codec::{CommunicationValue, DecodeError, DecodeLimits, EncodeLimits, TypeMap};
use mtp_common::CommunicationError;
#[cfg(feature = "pipes")]
use mtp_common::{FirstFrameDisposition, classify_first_frame};
#[cfg(feature = "pipes")]
use std::collections::HashSet;
use std::ops::Deref;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
@ -288,15 +292,15 @@ impl Sender {
Ok(Ok(())) => Ok(()),
Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
warn!("[Sender] write failed: peer sent STOP_SENDING (error code {code})");
Err(CommunicationError::StreamClosed)
Err(CommunicationError::DeliveryUnknown)
}
Ok(Err(other)) => {
warn!("[Sender] write failed: {other}");
Err(CommunicationError::StreamError)
Err(CommunicationError::DeliveryUnknown)
}
Err(_) => {
warn!("[Sender] write timed out (len={})", bytes.len());
Err(CommunicationError::StreamError)
Err(CommunicationError::DeliveryUnknown)
}
}
}
@ -398,15 +402,15 @@ impl Sender {
Ok(Ok(())) => Ok(()),
Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
warn!("[Sender] finish failed: peer sent STOP_SENDING (error code {code})");
Err(CommunicationError::StreamClosed)
Err(CommunicationError::DeliveryUnknown)
}
Ok(Err(other)) => {
warn!("[Sender] finish failed: {other}");
Err(CommunicationError::StreamError)
Err(CommunicationError::DeliveryUnknown)
}
Err(_) => {
warn!("[Sender] finish timed out");
Err(CommunicationError::StreamError)
Err(CommunicationError::DeliveryUnknown)
}
}
}
@ -745,6 +749,8 @@ struct ReceiverInner {
max_message_size: Arc<AtomicU64>,
type_map: Arc<RwLock<TypeMap>>,
decode_rejections: Arc<DecodeRejectionCounters>,
#[cfg(feature = "pipes")]
expected_pipes: Arc<std::sync::Mutex<HashSet<u32>>>,
}
impl Clone for Receiver {
@ -831,6 +837,10 @@ impl Receiver {
let accept_type_map = type_map.clone();
let decode_rejections = Arc::new(DecodeRejectionCounters::default());
let accept_decode_rejections = decode_rejections.clone();
#[cfg(feature = "pipes")]
let expected_pipes = Arc::new(std::sync::Mutex::new(HashSet::new()));
#[cfg(feature = "pipes")]
let accept_expected_pipes = expected_pipes.clone();
let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1)));
let accept_stream_limit = stream_limit.clone();
debug!(
@ -900,6 +910,8 @@ impl Receiver {
let stream_max_message_size = accept_max_message_size.clone();
let stream_type_map = accept_type_map.clone();
let stream_decode_rejections = accept_decode_rejections.clone();
#[cfg(feature = "pipes")]
let stream_expected_pipes = accept_expected_pipes.clone();
tokio::spawn(async move {
let _permit = permit;
@ -935,38 +947,54 @@ impl Receiver {
#[cfg(feature = "pipes")]
{
if msg.is_type(mtp_codec::CommunicationType::PipeRequest)
&& frame_count == 1
{
let Some(pipe_id) = msg.id().filter(|id| *id != 0) else {
let error = CommunicationError::Other(
"PipeRequest frame must contain a non-zero id".into(),
);
let _ = msg_tx_stream.send(Err(error.clone())).await;
stream_handle.close(Some(error));
if frame_count == 1 {
let is_pipe_request = msg.is_type(
mtp_codec::CommunicationType::PipeRequest,
);
let pipe_id = msg.id().filter(|id| *id != 0);
let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| {
stream_expected_pipes
.lock()
.is_ok_and(|mut expected| expected.remove(&pipe_id))
});
let disposition = match classify_first_frame(
is_pipe_request,
msg.id(),
pipe_is_expected,
) {
Ok(disposition) => disposition,
Err(error) => {
let _ = msg_tx_stream
.send(Err(error.clone()))
.await;
stream_handle.close(Some(error));
break;
}
};
if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
let description = msg
.get_str(mtp_codec::DataType::Description)
.unwrap_or("")
.to_string();
let pipe_reader = crate::pipe::PipeReader {
stream: s,
description,
pipe_id,
};
if pipe_tx_stream
.send(pipe_reader)
.await
.is_err()
{
stream_handle.close(Some(
CommunicationError::StreamClosed,
));
}
break;
};
let description = msg
.get_str(mtp_codec::DataType::Description)
.unwrap_or("")
.to_string();
let pipe_reader = crate::pipe::PipeReader {
stream: s,
description,
pipe_id,
};
if pipe_tx_stream
.send(pipe_reader)
.await
.is_err()
{
stream_handle.close(Some(
CommunicationError::StreamClosed,
));
}
break;
}
}
@ -1111,6 +1139,8 @@ impl Receiver {
max_message_size,
type_map,
decode_rejections,
#[cfg(feature = "pipes")]
expected_pipes,
}),
}
}
@ -1127,6 +1157,26 @@ impl Receiver {
*self.inner.type_map.write().await = type_map.clone();
}
#[cfg(feature = "pipes")]
pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
if pipe_id == 0 {
return Err(CommunicationError::Other("pipe id must be non-zero".into()));
}
self.inner
.expected_pipes
.lock()
.map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
.insert(pipe_id);
Ok(())
}
#[cfg(feature = "pipes")]
pub fn cancel_expected_pipe(&self, pipe_id: u32) {
if let Ok(mut expected) = self.inner.expected_pipes.lock() {
expected.remove(&pipe_id);
}
}
/// Return local counts for frames rejected by the structured decoder.
///
/// These counters are intentionally local-only; peers continue to receive

4
transport/src/framing.rs
View file

@ -82,6 +82,10 @@ mod tests {
async fn finish(&mut self) -> Result<(), CommunicationError> {
Ok(())
}
fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
Ok(())
}
}
#[tokio::test]

112
transport/src/generic_connection.rs
View file

@ -10,8 +10,12 @@ use crate::{
framing::{RetryClassifier, write_frame},
};
use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap};
use mtp_common::CommunicationError;
use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame};
#[cfg(feature = "pipes")]
use std::collections::HashSet;
use std::sync::Arc;
#[cfg(feature = "pipes")]
use std::sync::Mutex as StdMutex;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc};
use tokio::time::{Instant, timeout, timeout_at};
@ -85,9 +89,10 @@ impl<C: TransportConnection> GenericSender<C> {
)
.await
.map_err(|_| CommunicationError::StreamError)??;
timeout(self.policy.write_timeout, stream.finish())
.await
.map_err(|_| CommunicationError::StreamError)?
match timeout(self.policy.write_timeout, stream.finish()).await {
Ok(Ok(())) => Ok(()),
Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown),
}
}
crate::SendMode::PersistentStream => {
let mut stream = self.persistent.lock().await;
@ -204,6 +209,8 @@ pub struct GenericReceiver<C: TransportConnection> {
type_map: Arc<RwLock<TypeMap>>,
queue_notify: Arc<Notify>,
decode_rejections: Arc<DecodeRejectionCounters>,
#[cfg(feature = "pipes")]
expected_pipes: Arc<StdMutex<HashSet<u32>>>,
_accept_task: Arc<tokio::task::JoinHandle<()>>,
}
@ -219,6 +226,8 @@ impl<C: TransportConnection> Clone for GenericReceiver<C> {
type_map: self.type_map.clone(),
queue_notify: self.queue_notify.clone(),
decode_rejections: self.decode_rejections.clone(),
#[cfg(feature = "pipes")]
expected_pipes: self.expected_pipes.clone(),
_accept_task: self._accept_task.clone(),
}
}
@ -254,6 +263,10 @@ impl<C: TransportConnection> GenericReceiver<C> {
let task_queue_notify = queue_notify.clone();
let decode_rejections = Arc::new(DecodeRejectionCounters::default());
let task_decode_rejections = decode_rejections.clone();
#[cfg(feature = "pipes")]
let expected_pipes = Arc::new(StdMutex::new(HashSet::new()));
#[cfg(feature = "pipes")]
let task_expected_pipes = expected_pipes.clone();
let task_accept_task_tx = tx.clone();
#[cfg(feature = "pipes")]
let task_accept_task_pipe_tx = pipe_tx.clone();
@ -312,6 +325,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
let connection = task_connection.clone();
let type_map = task_type_map.clone();
let decode_rejections = task_decode_rejections.clone();
#[cfg(feature = "pipes")]
let expected_pipes = task_expected_pipes.clone();
tokio::spawn(async move {
let _permit = permit;
let mut stream = stream;
@ -443,37 +458,51 @@ impl<C: TransportConnection> GenericReceiver<C> {
#[cfg(feature = "pipes")]
{
if message.is_type(mtp_codec::CommunicationType::PipeRequest)
&& frames == 1
{
let Some(pipe_id) = message.id().filter(|id| *id != 0) else {
let error = CommunicationError::Other(
"PipeRequest frame must contain a non-zero id".into(),
);
let _ = tx.send(Err(error.clone())).await;
connection.close(
policy.application_close_code,
b"pipe request missing id",
);
break;
};
let description = message
.get_str(mtp_codec::DataType::Description)
.unwrap_or("")
.to_string();
let pipe_reader = PipeReader {
stream,
description,
pipe_id,
if frames == 1 {
let is_pipe_request =
message.is_type(mtp_codec::CommunicationType::PipeRequest);
let pipe_id = message.id().filter(|id| *id != 0);
let pipe_is_expected = is_pipe_request
&& pipe_id.is_some_and(|pipe_id| {
expected_pipes
.lock()
.is_ok_and(|mut expected| expected.remove(&pipe_id))
});
let disposition = match classify_first_frame(
is_pipe_request,
message.id(),
pipe_is_expected,
) {
Ok(disposition) => disposition,
Err(error) => {
let _ = tx.send(Err(error.clone())).await;
connection.close(
policy.application_close_code,
b"pipe request missing id",
);
break;
}
};
tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream");
if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
let description = message
.get_str(mtp_codec::DataType::Description)
.unwrap_or("")
.to_string();
if pipe_tx.send(pipe_reader).await.is_err() {
break;
let pipe_reader = PipeReader {
stream,
description,
pipe_id,
};
tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream");
if pipe_tx.send(pipe_reader).await.is_err() {
break;
}
return;
}
return;
}
}
@ -517,6 +546,8 @@ impl<C: TransportConnection> GenericReceiver<C> {
type_map,
queue_notify,
decode_rejections,
#[cfg(feature = "pipes")]
expected_pipes,
_accept_task: Arc::new(accept_task),
}
}
@ -524,6 +555,25 @@ impl<C: TransportConnection> GenericReceiver<C> {
*self.ping_sender.write().await = Some(sender);
}
#[cfg(feature = "pipes")]
pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
if pipe_id == 0 {
return Err(CommunicationError::Other("pipe id must be non-zero".into()));
}
self.expected_pipes
.lock()
.map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
.insert(pipe_id);
Ok(())
}
#[cfg(feature = "pipes")]
pub fn cancel_expected_pipe(&self, pipe_id: u32) {
if let Ok(mut expected) = self.expected_pipes.lock() {
expected.remove(&pipe_id);
}
}
/// Switch from the handshake frame limit to the application frame limit.
pub fn set_max_message_size(&self, max_message_size: u64) {
self.max_message_size

26
transport/src/transport_traits.rs
View file

@ -17,6 +17,7 @@ use mtp_common::CommunicationError;
pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError>;
async fn finish(&mut self) -> Result<(), CommunicationError>;
fn reset(&mut self, code: u32) -> Result<(), CommunicationError>;
}
/// A readable unidirectional stream suitable for MTP frames.
@ -28,6 +29,9 @@ pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
pub trait TransportRecvStream: tokio::io::AsyncRead + Send + Sync {
async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError>;
async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError>;
fn stop(self, code: u32) -> Result<(), CommunicationError>
where
Self: Sized;
}
/// A QUIC/WebTransport connection that provides MTP's unidirectional streams.
@ -47,7 +51,7 @@ impl TransportSendStream for wtransport::SendStream {
async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError> {
wtransport::SendStream::write_all(self, buf)
.await
.map_err(|_| CommunicationError::StreamError)
.map_err(|_| CommunicationError::DeliveryUnknown)
}
async fn finish(&mut self) -> Result<(), CommunicationError> {
@ -55,14 +59,23 @@ impl TransportSendStream for wtransport::SendStream {
.await
.map_err(|_| CommunicationError::StreamError)
}
fn reset(&mut self, code: u32) -> Result<(), CommunicationError> {
wtransport::SendStream::reset(self, wtransport::VarInt::from_u32(code))
.map_err(|_| CommunicationError::StreamClosed)
}
}
#[async_trait]
impl TransportRecvStream for wtransport::RecvStream {
async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> {
wtransport::RecvStream::read_exact(self, buf)
.await
.map_err(|_| CommunicationError::StreamError)
match wtransport::RecvStream::read_exact(self, buf).await {
Ok(()) => Ok(()),
Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => {
Err(CommunicationError::StreamClosed)
}
Err(_) => Err(CommunicationError::StreamError),
}
}
async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError> {
@ -76,6 +89,11 @@ impl TransportRecvStream for wtransport::RecvStream {
Err(_) => Err(CommunicationError::StreamError),
}
}
fn stop(self, code: u32) -> Result<(), CommunicationError> {
wtransport::RecvStream::stop(self, wtransport::VarInt::from_u32(code));
Ok(())
}
}
#[async_trait]

39
transport/tests/generic_pipe.rs
View file

@ -4,7 +4,7 @@ use async_trait::async_trait;
use mtp_codec::CommunicationValue;
use mtp_common::CommunicationError;
use mtp_transport::{
GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent,
GenericReceiver, GenericSender, Policy, SendMode, TransportConnection, TransportEvent,
TransportRecvStream, TransportSendStream,
};
use std::sync::Arc;
@ -53,6 +53,10 @@ impl TransportSendStream for MockSendStream {
.await
.map_err(|_| CommunicationError::StreamError)
}
fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
Ok(())
}
}
struct MockRecvStream {
@ -89,6 +93,10 @@ impl TransportRecvStream for MockRecvStream {
Err(_) => Err(CommunicationError::StreamError),
}
}
fn stop(self, _code: u32) -> Result<(), CommunicationError> {
Ok(())
}
}
#[derive(Clone)]
@ -157,6 +165,7 @@ async fn test_open_pipe_and_receive_reader() -> Result<(), Box<dyn std::error::E
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
receiver.expect_pipe(42)?;
let pipe_writer = sender.open_pipe(42, "test-pipe").await?;
let pipe_reader = receiver.receive_pipe().await?;
@ -174,6 +183,7 @@ async fn test_pipe_raw_data_roundtrip() -> Result<(), Box<dyn std::error::Error>
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
receiver.expect_pipe(1)?;
let mut pipe_writer = sender.open_pipe(1, "data-pipe").await?;
let data = b"hello through the pipe";
@ -195,6 +205,7 @@ async fn test_pipe_large_payload() -> Result<(), Box<dyn std::error::Error>> {
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
receiver.expect_pipe(7)?;
let mut pipe_writer = sender.open_pipe(7, "big-pipe").await?;
let data: Vec<u8> = (0..256 * 1024).map(|i| (i % 256) as u8).collect();
@ -223,6 +234,7 @@ async fn test_receive_event_dispatches_pipe() -> Result<(), Box<dyn std::error::
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
receiver.expect_pipe(99)?;
let mut pipe_writer = sender.open_pipe(99, "event-pipe").await?;
match receiver.receive_event().await? {
@ -261,22 +273,23 @@ async fn test_try_receive_pipe_returns_none_when_empty() -> Result<(), Box<dyn s
async fn test_regular_messages_still_work_alongside_pipes() -> Result<(), Box<dyn std::error::Error>>
{
let (conn_a, conn_b) = mock_connected_pair().await;
let policy = Arc::new(Policy::default());
let policy = Arc::new(Policy::default().with_send_mode(SendMode::SingleStreamPerMessage));
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest);
sender.send(&msg).await?;
let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
.with_id(1)
.add_typed_default(
mtp_codec::DataType::Description,
mtp_codec::DataValue::Str("mixed-pipe".into()),
);
sender.send(&request).await?;
let received = receiver.receive().await?;
assert_eq!(
received.get_type(),
mtp_codec::CommunicationType::BadRequest
.try_to_id(&mtp_codec::TypeMap::latest())
.unwrap()
);
assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest));
receiver.expect_pipe(1)?;
let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
let pipe_reader = receiver.receive_pipe().await?;
assert_eq!(pipe_reader.pipe_id(), 1);
@ -291,6 +304,8 @@ async fn test_multiple_pipes() -> Result<(), Box<dyn std::error::Error>> {
let sender = GenericSender::new(conn_a, policy.clone());
let receiver = GenericReceiver::new(conn_b, policy);
receiver.expect_pipe(10)?;
receiver.expect_pipe(20)?;
let mut pw1 = sender.open_pipe(10, "first").await?;
let mut pw2 = sender.open_pipe(20, "second").await?;