Prioritize handling inbound RPC calls to avoid high backpressure

Hopefully reducing congestion

Cargo.lock

@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "Inflector"
@@ -435,6 +435,12 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "bytecount"
+version = "0.6.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ce89b21cab1437276d2650d57e971f9d548a2d9037cc231abdc0562b97498ce"
+
 [[package]]
 name = "byteorder"
 version = "1.5.0"
@@ -626,6 +632,7 @@ dependencies = [
  "itoa",
  "rustversion",
  "ryu",
+ "serde",
  "static_assertions",
 ]
 
@@ -2021,6 +2028,7 @@ dependencies = [
  "miette 7.2.0",
  "percent-encoding",
  "quick-xml",
+ "rand",
  "reqwest",
  "rsa",
  "serde",
@@ -2047,11 +2055,13 @@ dependencies = [
 name = "magnetar_mmm_parser"
 version = "0.3.0-alpha"
 dependencies = [
+ "compact_str",
  "either",
  "emojis",
+ "nom",
+ "nom_locate",
  "quick-xml",
  "serde",
- "smallvec",
  "strum",
  "tracing",
  "unicode-segmentation",
@@ -2316,6 +2326,17 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "nom_locate"
+version = "4.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e3c83c053b0713da60c5b8de47fe8e494fe3ece5267b2f23090a07a053ba8f3"
+dependencies = [
+ "bytecount",
+ "memchr",
+ "nom",
+]
+
 [[package]]
 name = "nu-ansi-term"
 version = "0.46.0"
@@ -4177,6 +4198,7 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util",
 ]
 
 [[package]]

Cargo.toml

@@ -62,6 +62,7 @@ quick-xml = "0.36"
 redis = "0.26"
 regex = "1.9"
 rmp-serde = "1.3"
+rand = "0.8"
 rsa = "0.9"
 reqwest = "0.12"
 sea-orm = "1"
@@ -112,7 +113,7 @@ headers = { workspace = true }
 hyper = { workspace = true, features = ["full"] }
 reqwest = { workspace = true, features = ["hickory-dns"] }
 tokio = { workspace = true, features = ["full"] }
-tokio-stream = { workspace = true }
+tokio-stream = { workspace = true, features = ["full"] }
 tower = { workspace = true }
 tower-http = { workspace = true, features = ["cors", "trace", "fs"] }
 ulid = { workspace = true }

ext_federation/Cargo.toml

@@ -40,6 +40,7 @@ hyper = { workspace = true, features = ["full"] }
 percent-encoding = { workspace = true }
 reqwest = { workspace = true, features = ["stream", "hickory-dns"] }
 
+rand = { workspace = true }
 ed25519-dalek = { workspace = true, features = [
     "pem",
     "pkcs8",

ext_federation/src/ap_client.rs

@@ -4,6 +4,7 @@ use http::{HeaderMap, HeaderName, HeaderValue, Method};
 use indexmap::IndexSet;
 use serde_json::Value;
 use sha2::Digest;
+use std::fmt::Write;
 use std::{fmt::Display, string::FromUtf8Error, sync::Arc};
 use thiserror::Error;
 use tokio::task;
@@ -257,16 +258,17 @@ impl ApClientService for ApClientServiceDefaultProvider {
 
         let message = components
             .iter()
-            .map(|(k, v)| format!("{}: {}", k.as_ref(), v))
+            .fold(String::new(), |mut acc, (k, v)| {
-            .collect::<Vec<_>>()
+                writeln!(&mut acc, "{}: {}", k.as_ref(), v).unwrap();
-            .join("\n");
+                acc
+            });
 
         let key_id = signing_key.key_id.clone().into_owned();
         let key = signing_key.into_owned();
         let signature = task::spawn_blocking(move || {
             key
                 .key
-                .sign_base64(signing_algorithm, &message.into_bytes())
+                .sign_base64(signing_algorithm, message.trim_end().as_bytes())
         }).await??;
 
         Ok(ApSignature {
@@ -329,7 +331,7 @@ impl ApClientService for ApClientServiceDefaultProvider {
         }
 
         headers.insert(
-            HeaderName::from_lowercase(b"signature").unwrap(),
+            HeaderName::from_static("signature"),
             HeaderValue::try_from(signed.to_string())?,
         );
 
@@ -350,10 +352,9 @@ impl ApClientService for ApClientServiceDefaultProvider {
         signing_algorithm: SigningAlgorithm,
         expires: Option<chrono::DateTime<Utc>>,
         url: &str,
-        body: &Value,
+        body_bytes: Vec<u8>,
     ) -> Result<String, Self::Error> {
         let url = url.parse()?;
-        let body_bytes = serde_json::to_vec(body)?;
         // Move in, move out :3
         let (digest_raw, body_bytes) = task::spawn_blocking(move || {
             let mut sha = sha2::Sha256::new();
@@ -406,12 +407,12 @@ impl ApClientService for ApClientServiceDefaultProvider {
         }
 
         headers.insert(
-            HeaderName::from_lowercase(b"digest").unwrap(),
+            HeaderName::from_static("digest"),
             HeaderValue::try_from(digest_base64)?,
         );
 
         headers.insert(
-            HeaderName::from_lowercase(b"signature").unwrap(),
+            HeaderName::from_static("signature"),
             HeaderValue::try_from(signed.to_string())?,
         );
 

ext_federation/src/client/federation_client.rs

@@ -1,11 +1,11 @@
 use async_stream::stream;
-use futures_util::{select, stream::StreamExt, FutureExt, Stream, TryStreamExt};
+use futures_util::{stream::StreamExt, Stream, TryStreamExt};
 use headers::UserAgent;
 use hyper::body::Bytes;
 use reqwest::{redirect::Policy, Client, RequestBuilder};
 use serde_json::Value;
+use std::time::Duration;
 use thiserror::Error;
-use tokio::pin;
 use url::Url;
 
 use magnetar_core::web_model::ContentType;
@@ -58,6 +58,7 @@ impl FederationClient {
         let client = Client::builder()
             .https_only(force_https)
             .redirect(Policy::limited(5))
+            .timeout(Duration::from_secs(timeout_seconds))
             .build()?;
 
         Ok(FederationClient {
@@ -119,7 +120,7 @@ impl FederationRequestBuilder<'_> {
 
     async fn send_stream(
         self,
-    ) -> Result<impl Stream<Item = Result<Bytes, FederationClientError>>, FederationClientError>
+    ) -> Result<impl Stream<Item=Result<Bytes, FederationClientError>>, FederationClientError>
     {
         let mut body = self
             .builder
@@ -144,29 +145,13 @@ impl FederationRequestBuilder<'_> {
     }
 
     pub async fn send(self) -> Result<Vec<u8>, FederationClientError> {
-        let sleep = tokio::time::sleep(tokio::time::Duration::from_secs(
+        self.send_stream()
-            self.client.timeout_seconds,
+            .await?
-        ))
+            .try_fold(Vec::new(), |mut acc, b| async move {
-        .fuse();
+                acc.extend_from_slice(&b);
-        tokio::pin!(sleep);
+                Ok(acc)
-
+            })
-        let body = async move {
+            .await
-            self.send_stream()
-                .await?
-                .try_fold(Vec::new(), |mut acc, b| async move {
-                    acc.extend_from_slice(&b);
-                    Ok(acc)
-                })
-                .await
-        }
-        .fuse();
-
-        pin!(body);
-
-        select! {
-            b = body => b,
-            _ = sleep => Err(FederationClientError::TimeoutError)
-        }
     }
 
     pub async fn json(self) -> Result<Value, FederationClientError> {

ext_federation/src/crypto.rs

@@ -2,7 +2,7 @@ use rsa::pkcs1::DecodeRsaPrivateKey;
 use rsa::pkcs1::DecodeRsaPublicKey;
 use rsa::pkcs8::DecodePrivateKey;
 use rsa::pkcs8::DecodePublicKey;
-use rsa::signature::Verifier;
+use rsa::signature::{RandomizedSigner, Verifier};
 use rsa::{
     sha2::{Sha256, Sha512},
     signature::Signer,
@@ -268,10 +268,10 @@ impl ApHttpSigningKey<'_> {
     ) -> Result<Vec<u8>, ApSigningError> {
         match (self, algorithm) {
             (Self::RsaSha256(key), SigningAlgorithm::RsaSha256 | SigningAlgorithm::Hs2019) => {
-                Ok(Box::<[u8]>::from(key.sign(message)).into_vec())
+                Ok(Box::<[u8]>::from(key.sign_with_rng(&mut rand::thread_rng(), message)).into_vec())
             }
             (Self::RsaSha512(key), SigningAlgorithm::Hs2019) => {
-                Ok(Box::<[u8]>::from(key.sign(message)).into_vec())
+                Ok(Box::<[u8]>::from(key.sign_with_rng(&mut rand::thread_rng(), message)).into_vec())
             }
             (Self::Ed25519(key), SigningAlgorithm::Hs2019) => {
                 Ok(key.sign(message).to_bytes().to_vec())

ext_federation/src/lib.rs

@@ -171,6 +171,6 @@ pub trait ApClientService: Send + Sync {
         signing_algorithm: SigningAlgorithm,
         expires: Option<chrono::DateTime<Utc>>,
         url: &str,
-        body: &Value,
+        body: Vec<u8>,
     ) -> Result<String, Self::Error>;
 }

fe_calckey/frontend/client/src/components/MagNoteHeader.vue

@@ -74,10 +74,10 @@ const showTicker =
     display: flex;
     align-items: center;
     white-space: nowrap;
-    justify-self: flex-end;
     border-radius: 100px;
     font-size: 0.8em;
     text-shadow: 0 2px 2px var(--shadow);
+    gap: 0 0.1em;
 
     > .avatar {
         width: 3.7em;
@@ -99,6 +99,8 @@ const showTicker =
                 overflow: hidden;
                 text-overflow: ellipsis;
                 gap: 0.1em 0;
+                display: flex;
+                flex-wrap: wrap;
             }
 
             &:last-child {
@@ -151,7 +153,6 @@ const showTicker =
             margin: 0 0.5em 0 0;
             overflow: hidden;
             text-overflow: ellipsis;
-            align-self: flex-start;
             font-size: 0.9em;
         }
 

fe_calckey/frontend/client/src/components/MkInstanceTicker.vue

@@ -4,7 +4,7 @@
         v-tooltip="
             capitalize(
                 magTransProperty(instance, 'software_name', 'softwareName') ??
-                    '?',
+                    '?'
             )
         "
         ref="ticker"
@@ -40,7 +40,7 @@ const instance = props.instance ?? {
     name: instanceName,
     themeColor: (
         document.querySelector(
-            'meta[name="theme-color-orig"]',
+            'meta[name="theme-color-orig"]'
         ) as HTMLMetaElement
     )?.content,
     softwareName: (Instance.softwareName || "Magnetar") as string | null,
@@ -61,18 +61,18 @@ const bg = {
 };
 
 function getInstanceIcon(
-    instance?: Misskey.entities.User["instance"] | types.InstanceTicker | null,
+    instance?: Misskey.entities.User["instance"] | types.InstanceTicker | null
 ): string {
     if (!instance) return "/client-assets/dummy.png";
 
     return (
         getProxiedImageUrlNullable(
             magTransProperty(instance, "icon_url", "iconUrl"),
-            "preview",
+            "preview"
         ) ??
         getProxiedImageUrlNullable(
             magTransProperty(instance, "favicon_url", "faviconUrl"),
-            "preview",
+            "preview"
         ) ??
         "/client-assets/dummy.png"
     );
@@ -90,6 +90,7 @@ function getInstanceIcon(
     font-size: 0.8em;
     text-shadow: 0 2px 2px var(--shadow);
     overflow: hidden;
+
     .header > .body & {
         width: max-content;
         max-width: 100%;
@@ -108,11 +109,9 @@ function getInstanceIcon(
         font-weight: bold;
         text-overflow: ellipsis;
         white-space: nowrap;
-        text-shadow:
+        text-shadow: -1px -1px 0 var(--bg), 1px -1px 0 var(--bg),
-            -1px -1px 0 var(--bg),
+            -1px 1px 0 var(--bg), 1px 1px 0 var(--bg);
-            1px -1px 0 var(--bg),
+
-            -1px 1px 0 var(--bg),
-            1px 1px 0 var(--bg);
         .article > .main &,
         .header > .body & {
             display: unset;

magnetar_mmm_parser/Cargo.toml

@@ -11,8 +11,10 @@ xml = ["dep:quick-xml"]
 [dependencies]
 either = { workspace = true }
 emojis = { workspace = true }
+nom = { workspace = true }
+nom_locate = { workspace = true }
+compact_str = { workspace = true, features = ["serde"] }
 serde = { workspace = true, features = ["derive"] }
-smallvec = { workspace = true }
 strum = { workspace = true, features = ["derive"] }
 tracing = { workspace = true }
 unicode-segmentation = { workspace = true }

magnetar_mmm_parser/src/output_types.rs

@@ -1,261 +0,0 @@
-use either::Either;
-use serde::{Deserialize, Serialize};
-use std::borrow::Cow;
-use std::collections::HashMap;
-use strum::IntoStaticStr;
-
-#[derive(Debug, Clone, Deserialize, Serialize, Eq, PartialEq)]
-pub enum Token<'a> {
-    PlainText(Cow<'a, str>),
-    Sequence(Vec<Token<'a>>),
-    Quote(Vec<Token<'a>>),
-    Small(Vec<Token<'a>>),
-    BoldItalic(Vec<Token<'a>>),
-    Bold(Vec<Token<'a>>),
-    Italic(Vec<Token<'a>>),
-    Center(Vec<Token<'a>>),
-    Strikethrough(Vec<Token<'a>>),
-    PlainTag(String),
-    InlineCode(String),
-    InlineMath(String),
-    UrlRaw(String),
-    UrlNoEmbed(String),
-    Link {
-        label: Vec<Token<'a>>,
-        href: String,
-    },
-    LinkNoEmbed {
-        label: Vec<Token<'a>>,
-        href: String,
-    },
-    BlockCode {
-        lang: Option<String>,
-        inner: String,
-    },
-    BlockMath(String),
-    Function {
-        name: String,
-        params: HashMap<String, Option<String>>,
-        inner: Vec<Token<'a>>,
-    },
-    Mention {
-        name: String,
-        host: Option<String>,
-        mention_type: MentionType,
-    },
-    UnicodeEmoji(String),
-    ShortcodeEmoji {
-        shortcode: String,
-        host: Option<String>,
-    },
-    Hashtag(String),
-}
-
-#[derive(Copy, Clone, Debug, Eq, PartialEq, Deserialize, Serialize, IntoStaticStr)]
-// The alternative would be to implement a serde serializer for this one enum, but that's disgusting
-#[strum(serialize_all = "snake_case")]
-#[serde(rename_all = "snake_case")]
-pub enum MentionType {
-    Community,
-    User,
-    MatrixUser,
-}
-
-impl MentionType {
-    pub fn to_char(&self) -> char {
-        match self {
-            MentionType::Community => '!',
-            MentionType::User => '@',
-            MentionType::MatrixUser => ':',
-        }
-    }
-
-    pub fn separator(&self) -> char {
-        match self {
-            MentionType::Community | MentionType::User => '@',
-            MentionType::MatrixUser => ':',
-        }
-    }
-}
-
-
-impl Token<'_> {
-    fn str_content_left(&self) -> Option<&str> {
-        match self {
-            Token::PlainText(text) => Some(text.as_ref()),
-            Token::Sequence(tokens) => tokens.first().and_then(Token::str_content_left),
-            Token::Quote(inner) => inner.str_content_left(),
-            Token::Small(inner) => inner.str_content_left(),
-            Token::Bold(inner) => inner.str_content_left(),
-            Token::Italic(inner) => inner.str_content_left(),
-            Token::Center(inner) => inner.str_content_left(),
-            Token::Strikethrough(inner) => inner.str_content_left(),
-            Token::PlainTag(tag) => Some(tag.as_ref()),
-            Token::UrlRaw(url) => Some(url.as_ref()),
-            Token::UrlNoEmbed(url) => Some(url.as_ref()),
-            Token::Link { label, .. } => label.str_content_left(),
-            Token::Function { inner, .. } => inner.str_content_left(),
-            Token::Mention { name, .. } => Some(name.as_ref()),
-            Token::UnicodeEmoji(code) => Some(code.as_ref()),
-            Token::Hashtag(tag) => Some(tag.as_ref()),
-            _ => None,
-        }
-    }
-
-    fn str_content_right(&self) -> Option<&str> {
-        match self {
-            Token::PlainText(text) => Some(text.as_ref()),
-            Token::Sequence(tokens) => tokens.last().and_then(Token::str_content_right),
-            Token::Quote(inner) => inner.str_content_right(),
-            Token::Small(inner) => inner.str_content_right(),
-            Token::Bold(inner) => inner.str_content_right(),
-            Token::Italic(inner) => inner.str_content_right(),
-            Token::Center(inner) => inner.str_content_right(),
-            Token::Strikethrough(inner) => inner.str_content_right(),
-            Token::PlainTag(tag) => Some(tag.as_ref()),
-            Token::UrlRaw(url) => Some(url.as_ref()),
-            Token::UrlNoEmbed(url) => Some(url.as_ref()),
-            Token::Link { label, .. } => label.str_content_right(),
-            Token::Function { inner, .. } => inner.str_content_right(),
-            Token::Mention { name, .. } => Some(name.as_ref()),
-            Token::UnicodeEmoji(code) => Some(code.as_ref()),
-            Token::Hashtag(tag) => Some(tag.as_ref()),
-            _ => None,
-        }
-    }
-
-    fn inner(&self) -> Token {
-        match self {
-            plain @ Token::PlainText(_) => plain.clone(),
-            sequence @ Token::Sequence(_) => sequence.clone(),
-            Token::Quote(inner) => inner.inner(),
-            Token::Small(inner) => inner.inner(),
-            Token::Bold(inner) => inner.inner(),
-            Token::Italic(inner) => inner.inner(),
-            Token::Center(inner) => inner.inner(),
-            Token::Strikethrough(inner) => inner.inner(),
-            Token::PlainTag(text) => Token::PlainText(text.clone().into()),
-            Token::InlineCode(code) => Token::PlainText(code.clone().into()),
-            Token::InlineMath(math) => Token::PlainText(math.clone().into()),
-            Token::UrlRaw(url) => Token::PlainText(url.clone().into()),
-            Token::UrlNoEmbed(url) => Token::PlainText(url.clone().into()),
-            Token::Link { label, .. } => label.inner(),
-            Token::BlockCode { inner, .. } => Token::PlainText(inner.clone().into()),
-            Token::BlockMath(math) => Token::PlainText(math.clone().into()),
-            Token::Function { inner, .. } => inner.inner(),
-            Token::Mention { name, .. } => Token::PlainText(name.clone().into()),
-            Token::UnicodeEmoji(code) => Token::PlainText(code.clone().into()),
-            Token::ShortcodeEmoji { shortcode, .. } => Token::PlainText(shortcode.clone().into()),
-            Token::Hashtag(tag) => Token::PlainText(tag.clone().into()),
-        }
-    }
-
-    fn merged(&self) -> Token {
-        match self {
-            Token::Sequence(tokens) => {
-                let tokens_multi = tokens.iter().fold(Vec::new(), |mut acc, tok| {
-                    if let Some(Token::PlainText(last)) = acc.last_mut() {
-                        if let Token::PlainText(tok_text) = tok {
-                            *last += tok_text.as_ref();
-
-                            return acc;
-                        }
-                    }
-
-                    if let Token::Sequence(seq) = tok {
-                        let items = seq.iter().map(Token::merged).flat_map(|t| match t {
-                            Token::Sequence(seq) => Either::Left(seq.into_iter()),
-                            other => Either::Right(std::iter::once(other)),
-                        });
-
-                        for item in items {
-                            if let Some(Token::PlainText(last)) = acc.last_mut() {
-                                if let Token::PlainText(tok_text) = item {
-                                    *last += tok_text.as_ref();
-
-                                    continue;
-                                }
-                            }
-
-                            acc.push(item);
-                        }
-
-                        return acc;
-                    }
-
-                    acc.push(tok.merged());
-                    acc
-                });
-
-                if tokens_multi.len() == 1 {
-                    return tokens_multi.into_iter().next().unwrap();
-                }
-
-                Token::Sequence(tokens_multi)
-            }
-            Token::Quote(inner) => Token::Quote(Box::new(inner.merged())),
-            Token::Small(inner) => Token::Small(Box::new(inner.merged())),
-            Token::Bold(inner) => Token::Bold(Box::new(inner.merged())),
-            Token::Italic(inner) => Token::Italic(Box::new(inner.merged())),
-            Token::Center(inner) => Token::Center(Box::new(inner.merged())),
-            Token::Strikethrough(inner) => Token::Strikethrough(Box::new(inner.merged())),
-            Token::Link { label, href } => Token::Link {
-                label: Box::new(label.merged()),
-                href: href.clone(),
-            },
-            Token::LinkNoEmbed { label, href } => Token::LinkNoEmbed {
-                label: Box::new(label.merged()),
-                href: href.clone(),
-            },
-            Token::Function {
-                name,
-                params,
-                inner,
-            } => Token::Function {
-                name: name.clone(),
-                params: params.clone(),
-                inner: Box::new(inner.merged()),
-            },
-            other => other.clone(),
-        }
-    }
-
-    pub fn walk_map_collect<T>(&self, func: &impl Fn(&Token) -> Option<T>, out: &mut Vec<T>) {
-        if let Some(v) = func(self) {
-            out.push(v)
-        }
-
-        match self {
-            Token::Sequence(items) => {
-                items.iter().for_each(|tok| tok.walk_map_collect(func, out));
-            }
-            Token::Quote(inner)
-            | Token::Small(inner)
-            | Token::Bold(inner)
-            | Token::Italic(inner)
-            | Token::Center(inner)
-            | Token::Function { inner, .. }
-            | Token::Link { label: inner, .. }
-            | Token::Strikethrough(inner) => inner.walk_map_collect(func, out),
-            _ => {}
-        }
-    }
-
-    pub fn walk_speech_transform(&mut self, func: &impl Fn(&mut Cow<'_, str>)) {
-        match self {
-            Token::Sequence(items) => {
-                items
-                    .iter_mut()
-                    .for_each(|tok| tok.walk_speech_transform(func));
-            }
-            Token::Small(inner)
-            | Token::Bold(inner)
-            | Token::Italic(inner)
-            | Token::Center(inner)
-            | Token::Function { inner, .. }
-            | Token::Strikethrough(inner) => inner.walk_speech_transform(func),
-            Token::PlainText(text) => func(text),
-            _ => {}
-        }
-    }
-}

magnetar_mmm_parser/src/parser.rs

@@ -1,157 +0,0 @@
-use crate::types::{Effect, Input, Parser, ParserCont, ParserRet, State};
-
-fn line_start<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {
-    match inp.view().as_bytes() {
-        [b'>', b' ', ..] => cont.continue_with2((line_start, quote)),
-        [b'`', b'`', b'`', ..] => cont.continue_with(CodeBlock {}),
-        [b'\\', b'[', ..] => cont.continue_with(BlockMath {}),
-        [b'<', b'c', b'e', b'n', b't', b'e', b'r', b'>', ..] => cont.continue_with2((inline, center_tag_end)),
-        _ => cont.continue_with(inline)
-    }
-}
-
-fn inline<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {
-    match inp.view().as_bytes() {
-        [b'\n', ..] => return cont.continue_with(line_start),
-        [b'<', b'b', b'>', ..] => return cont.continue_with(inline),
-        [b'<', b's', b'>', ..] => return cont.continue_with(inline),
-        [b'<', b'i', b'>', ..] => return cont.continue_with(inline),
-        [b'<', b'p', b'l', b'a', b'i', b'n', b'>', ..] => return cont.continue_with(inline),
-        [b'<', b's', b'm', b'a', b'l', b'l', b'>', ..] => return cont.continue_with(inline),
-        [b'*', b'*', ..] => return cont.continue_with(inline),
-        [b'_', b'_', ..] => return cont.continue_with(inline),
-        [b'*', ..] => return cont.continue_with(inline),
-        [b'_', ..] => return cont.continue_with(inline),
-        [b'~', b'~', ..] => return cont.continue_with(inline),
-        [b'`', ..] => return cont.continue_with(inline),
-        [b'\\', b'(', ..] => return cont.continue_with(inline),
-    };
-}
-
-fn text_or_emoji<'a>(
-    state: &mut State,
-    input: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {
-    let Some(view) = input.next() else {
-        return;
-    };
-
-    let emoji_str = view.trim_end_matches(['\u{200c}', '\u{200d}']);
-    if let Some(_) = emojis::get(emoji_str) {
-        output(Effect::Output(emoji_str));
-        return;
-    };
-
-    output(Effect::Output(view));
-}
-
-fn block_quote_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-fn code_block_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-fn block_math_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-
-fn center_tag_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-
-#[derive(Copy, Clone)]
-enum TagInlineKind {
-    TagSmall,
-    TagPlain,
-    TagBold,
-    TagItalic,
-    TagStrikethrough,
-}
-
-struct TagInline {
-    kind: TagInlineKind,
-}
-
-impl Parser for TagInline {}
-
-
-fn inline_math_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-
-fn inline_code_end<'a>(
-    state: &mut State,
-    inp: &mut impl Input<'a>,
-    _output: &'_ mut impl FnMut(Effect<'a>),
-    cont: impl ParserCont,
-) -> ParserRet {}
-
-
-struct Url {}
-
-impl Parser for Url {
-    fn take<'a>(
-        &mut self,
-        state: State,
-        input: &mut impl Input<'a>,
-        output: &'_ mut impl FnMut(Effect<'a>),
-    ) -> impl Parser {}
-}
-
-#[inline]
-fn url_chars_base<'a>(&self, input: Span<'a>) -> IResult<Span<'a>, Span<'a>> {
-    alt((
-        recognize(tuple((
-            tag("["),
-            many_till(
-                self.increase_nesting(self.partial_span(Self::url_chars_base)),
-                tag("]"),
-            ),
-        ))),
-        recognize(tuple((
-            tag("("),
-            many_till(
-                self.increase_nesting(self.partial_span(Self::url_chars_base)),
-                tag(")"),
-            ),
-        ))),
-        recognize(tuple((
-            not(satisfy(char::is_control)),
-            not(satisfy(char::is_whitespace)),
-            not(one_of(")]>")),
-            anychar,
-        ))),
-    ))(input)
-}

magnetar_mmm_parser/src/test.rs

@@ -1,762 +0,0 @@
-#![cfg(test)]
-use std::collections::HashMap;
-
-use crate::output_types::{MentionType, Token};
-use crate::{parse_full, xml_write::to_xml_string};
-
-#[test]
-fn parse_empty() {
-    assert_eq!(parse_full(""), Token::Sequence(vec![]));
-}
-
-#[test]
-fn parse_url_chars() {
-    let ctx = Context::default();
-
-    assert_eq!(
-        ctx.url_chars(tag(")"), true)(Span::new_extra(
-            "https://en.wikipedia.org/wiki/Sandbox_(computer_security))",
-            SpanMeta::default(),
-        ))
-            .unwrap()
-            .1
-            .into_fragment(),
-        "https://en.wikipedia.org/wiki/Sandbox_(computer_security)"
-    );
-
-    assert_eq!(
-        ctx.url_chars(tag(")"), true)(Span::new_extra(
-            "https://en.wikipedia.org/wiki/Sandbox_(computer_security)))",
-            SpanMeta::default()
-        ))
-        .unwrap()
-        .1
-        .into_fragment(),
-        "https://en.wikipedia.org/wiki/Sandbox_(computer_security)",
-    );
-
-    assert_eq!(
-        ctx.url_chars(tag(")"), true)(Span::new_extra(
-            "https://cs.wikipedia.org/wiki/Among_Us  ",
-            SpanMeta::default()
-        ))
-        .unwrap()
-        .1
-        .into_fragment(),
-        "https://cs.wikipedia.org/wiki/Among_Us",
-    );
-
-    assert_eq!(
-        ctx.url_chars(tag(")"), true)(Span::new_extra(
-            "https://cs.wikipedia.org/wiki/Among Us  )",
-            SpanMeta::default(),
-        ))
-            .unwrap()
-            .1
-            .into_fragment(),
-        "https://cs.wikipedia.org/wiki/Among Us"
-    );
-
-    assert_eq!(
-        ctx.url_chars(tag(")"), false)(Span::new_extra(
-            "https://en.wikipedia.org/wiki/Among Us  )",
-            SpanMeta::default(),
-        ))
-            .unwrap()
-            .1
-            .into_fragment(),
-        "https://en.wikipedia.org/wiki/Among"
-    );
-}
-
-#[test]
-fn parse_formatting() {
-    assert_eq!(
-        parse_full(r#"~~stikethrough~~"#),
-        Token::Strikethrough(vec![Token::PlainText("stikethrough".into())]),
-    );
-
-    assert_eq!(
-        parse_full(r#"**bold**"#),
-        Token::Bold(vec![Token::PlainText("bold".into())]),
-    );
-
-    assert_eq!(
-        parse_full(r#"*italic*"#),
-        Token::Italic(vec![Token::PlainText("italic".into())]),
-    );
-
-    assert_eq!(
-        parse_full(r#"* italic *"#),
-        Token::PlainText("* italic *".into())
-    );
-
-    assert_eq!(
-        parse_full("snake_case_variable"),
-        Token::PlainText("snake_case_variable".into())
-    );
-
-    assert_eq!(
-        parse_full("intra*word*italic"),
-        Token::Sequence(vec![
-            Token::PlainText("intra".into()),
-            Token::Italic(vec![Token::PlainText("word".into())]),
-            Token::PlainText("italic".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"_ italic *"#),
-        Token::PlainText("_ italic *".into())
-    );
-
-    assert_eq!(
-        parse_full(r#"long text with a *footnote <b>text</b>"#),
-        Token::Sequence(vec![
-            Token::PlainText("long text with a *footnote ".into()),
-            Token::Bold(vec![Token::PlainText("text".into())]),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"*"italic"*"#),
-        Token::Italic(vec![Token::PlainText("\"italic\"".into())])
-    );
-
-    assert_eq!(
-        parse_full(r#"not code `code` also not code"#),
-        Token::Sequence(vec![
-            Token::PlainText("not code ".into()),
-            Token::InlineCode("code".into()),
-            Token::PlainText(" also not code".into())
-        ]),
-    );
-
-    assert_eq!(
-        parse_full(r#"not code `code` also `not code"#),
-        Token::Sequence(vec![
-            Token::PlainText("not code ".into()),
-            Token::InlineCode("code".into()),
-            Token::PlainText(" also `not code".into())
-        ]),
-    );
-
-    assert_eq!(
-        parse_full(r#"not code `*not bold*` also not code"#),
-        Token::Sequence(vec![
-            Token::PlainText("not code ".into()),
-            Token::InlineCode("*not bold*".into()),
-            Token::PlainText(" also not code".into())
-        ]),
-    );
-
-    assert_eq!(
-        parse_full(r#"***bold italic***"#),
-        Token::Bold(vec![Token::Italic(vec![Token::PlainText(
-            "bold italic".into()
-        )])])
-    );
-
-    assert_eq!(
-        parse_full(r#"<b><i>bold italic</i></b>"#),
-        Token::Bold(vec![Token::Italic(vec![Token::PlainText(
-            "bold italic".into()
-        )])])
-    );
-
-    assert_eq!(
-        parse_full("~~*hello\nworld*"),
-        Token::Sequence(vec![
-            Token::PlainText("~~".into()),
-            Token::Italic(vec![Token::PlainText("hello\nworld".into())]),
-        ])
-    )
-}
-
-#[test]
-fn parse_flanking() {
-    assert_eq!(
-        parse_full(r#"aaa*iii*bbb"#),
-        Token::Sequence(vec![
-            Token::PlainText("aaa".into()),
-            Token::Italic(vec![Token::PlainText("iii".into())]),
-            Token::PlainText("bbb".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"aaa_nnn_bbb"#),
-        Token::PlainText("aaa_nnn_bbb".into())
-    );
-
-    assert_eq!(
-        parse_full("aaa\n_iii_\nbbb"),
-        Token::Sequence(vec![
-            Token::PlainText("aaa\n".into()),
-            Token::Italic(vec![Token::PlainText("iii".into())]),
-            Token::PlainText("\nbbb".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"*iii*"#),
-        Token::Italic(vec![Token::PlainText("iii".into())])
-    );
-
-    assert_eq!(
-        parse_full(r#"_iii_"#),
-        Token::Italic(vec![Token::PlainText("iii".into())])
-    );
-
-    assert_eq!(
-        parse_full(r#"aaa*iii*"#),
-        Token::Sequence(vec![
-            Token::PlainText("aaa".into()),
-            Token::Italic(vec![Token::PlainText("iii".into())]),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"*iii*bbb"#),
-        Token::Sequence(vec![
-            Token::Italic(vec![Token::PlainText("iii".into())]),
-            Token::PlainText("bbb".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(r#"aaa_nnn_"#),
-        Token::PlainText("aaa_nnn_".into())
-    );
-
-    assert_eq!(
-        parse_full(r#"_nnn_bbb"#),
-        Token::PlainText("_nnn_bbb".into())
-    );
-}
-
-#[test]
-fn parse_long() {
-    parse_full(&"A".repeat(20000));
-
-    parse_full(&"*A".repeat(20000));
-
-    parse_full(&"@A".repeat(20000));
-}
-
-#[test]
-fn parse_complex() {
-    assert_eq!(
-        parse_full(r"\( nya^3 \)"),
-        Token::InlineMath(" nya^3 ".to_string())
-    );
-
-    assert_eq!(
-        parse_full("\\( nya^3 \n \\)"),
-        Token::PlainText("\\( nya^3 \n \\)".into())
-    );
-
-    assert_eq!(
-        parse_full(r"`AbstractProxyFactoryBean`"),
-        Token::InlineCode("AbstractProxyFactoryBean".to_string())
-    );
-
-    assert_eq!(
-        parse_full("`let x = \n 5;`"),
-        Token::PlainText("`let x = \n 5;`".into())
-    );
-
-    assert_eq!(
-        parse_full(
-            r#"
-```js
-var x = undefined;
-```"#
-        ),
-        Token::BlockCode {
-            lang: Some("js".to_string()),
-            inner: "var x = undefined;".to_string(),
-        }
-    );
-
-    assert_eq!(
-        parse_full(
-            r"
-\[
-a^2 + b^2 = c^2
-\]"
-        ),
-        Token::BlockMath("a^2 + b^2 = c^2".to_string())
-    );
-
-    assert_eq!(
-        parse_full(r"\[ x^2 + y^2 = z^2 \]"),
-        Token::BlockMath("x^2 + y^2 = z^2".to_string())
-    );
-
-    assert_eq!(
-        parse_full(
-            r#"<center>centered
-    🦋🏳️‍⚧️
-    text</center>"#
-        ),
-        Token::Center(vec![
-            Token::PlainText("centered\n".into()),
-            Token::UnicodeEmoji("🦋".into()),
-            Token::UnicodeEmoji("🏳️‍⚧️".into()),
-            Token::PlainText("\ntext".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full(
-            r#"> <center>centered
-> 👩🏽‍🤝‍👩🏼
-> text</center>"#
-        ),
-        Token::Quote(vec![Token::Center(vec![
-            Token::PlainText("centered\n".into()),
-            Token::UnicodeEmoji("👩🏽‍🤝‍👩🏼".into()),
-            Token::PlainText("\ntext".into())
-        ])]),
-    );
-
-    assert_eq!(
-        parse_full(r#"$[x2 $[sparkle 🥺]💜$[spin.y,speed=5s ❤️]🦊]"#),
-        Token::Function {
-            name: "x2".into(),
-            params: HashMap::new(),
-            inner: vec![
-                Token::Function {
-                    name: "sparkle".into(),
-                    params: HashMap::new(),
-                    inner: vec![Token::UnicodeEmoji("🥺".into())],
-                },
-                Token::UnicodeEmoji("💜".into()),
-                Token::Function {
-                    name: "spin".into(),
-                    params: {
-                        let mut params = HashMap::new();
-                        params.insert("y".into(), None);
-                        params.insert("speed".into(), Some("5s".into()));
-                        params
-                    },
-                    inner: vec![Token::UnicodeEmoji("❤️".into())],
-                },
-                Token::UnicodeEmoji("🦊".into()),
-            ]
-        },
-    );
-
-    assert_eq!(
-        parse_full(r#"<b>bold @tag1 <i> @tag2 </b>italic</i>"#),
-        Token::Sequence(vec![
-            Token::PlainText("<b>bold ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag1".into(),
-                host: None
-            },
-            Token::PlainText(" <i> ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag2".into(),
-                host: None
-            },
-            Token::PlainText(" </b>italic</i>".into())
-        ]),
-    );
-
-    assert_eq!(
-        parse_full(
-            r#"
-> test
-> <i>
-> italic
-> </i>
->> Nested quote
-"#
-        ),
-        Token::Quote(vec![
-            Token::PlainText("test\n".into()),
-            Token::Italic(vec![Token::PlainText("\nitalic\n".into())]),
-            Token::Quote(vec![Token::PlainText("Nested quote".into())])
-        ]),
-    );
-}
-
-#[test]
-fn parse_link() {
-    assert_eq!(
-        parse_full("IPv4 test: <https://0>"),
-        Token::Sequence(vec![
-            Token::PlainText("IPv4 test: ".into()),
-            Token::UrlNoEmbed("https://0".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("IPv4 test: <https://127.0.0.1>"),
-        Token::Sequence(vec![
-            Token::PlainText("IPv4 test: ".into()),
-            Token::UrlNoEmbed("https://127.0.0.1".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("IPv6 test: <https://[::2f:1]/nya>"),
-        Token::Sequence(vec![
-            Token::PlainText("IPv6 test: ".into()),
-            Token::UrlNoEmbed("https://[::2f:1]/nya".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("IPv6 test: https://[::2f:1]/nya"),
-        Token::Sequence(vec![
-            Token::PlainText("IPv6 test: ".into()),
-            Token::UrlRaw("https://[::2f:1]/nya".into()),
-        ])
-    );
-
-    // IDNs
-    assert_eq!(
-        parse_full("IDN test: https://www.háčkyčárky.cz/"),
-        Token::Sequence(vec![
-            Token::PlainText("IDN test: ".into()),
-            Token::UrlRaw("https://www.háčkyčárky.cz/".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("Link test: [label](https://example.com)"),
-        Token::Sequence(vec![
-            Token::PlainText("Link test: ".into()),
-            Token::Link {
-                label: vec![Token::PlainText("label".into())],
-                href: "https://example.com".into()
-            },
-        ])
-    );
-
-    assert_eq!(
-        parse_full("test #hashtag tail"),
-        Token::Sequence(vec![
-            Token::PlainText("test ".into()),
-            Token::Hashtag("hashtag".into()),
-            Token::PlainText(" tail".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("not#hashtag tail"),
-        Token::PlainText("not#hashtag tail".into())
-    );
-
-    assert_eq!(
-        parse_full("<https://example.com>"),
-        Token::UrlNoEmbed("https://example.com".into())
-    );
-
-    // Adjacent links okay
-    assert_eq!(
-        parse_full("<https://example.com/><https://awawa.gay/>"),
-        Token::Sequence(vec![
-            Token::UrlNoEmbed("https://example.com/".into()),
-            Token::UrlNoEmbed("https://awawa.gay/".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("Link test: ?[label](https://awawa.gay)"),
-        Token::Sequence(vec![
-            Token::PlainText("Link test: ".into()),
-            Token::LinkNoEmbed {
-                label: vec![Token::PlainText("label".into())],
-                href: "https://awawa.gay".into(),
-            },
-        ])
-    );
-
-    assert_eq!(
-        parse_full("Link test: ?[label](https://awawa.gay)test"),
-        Token::Sequence(vec![
-            Token::PlainText("Link test: ".into()),
-            Token::LinkNoEmbed {
-                label: vec![Token::PlainText("label".into())],
-                href: "https://awawa.gay".into(),
-            },
-            Token::PlainText("test".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("Link test: (?[label](https://awawa.gay))"),
-        Token::Sequence(vec![
-            Token::PlainText("Link test: (".into()),
-            Token::LinkNoEmbed {
-                label: vec![Token::PlainText("label".into())],
-                href: "https://awawa.gay".into(),
-            },
-            Token::PlainText(")".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("Link test: ?[label](https://awawa.gay"), // Missing closing bracket
-        Token::Sequence(vec![
-            Token::PlainText("Link test: ?[label](".into()),
-            Token::UrlRaw("https://awawa.gay".into()),
-        ])
-    );
-}
-
-#[test]
-fn limit_nesting() {
-    let mut tok = Token::PlainText(" <s><i>test</i></s> ".into());
-    for _ in 0..DEFAULT_DEPTH_LIMIT {
-        tok = Token::Bold(Box::new(tok));
-    }
-
-    assert_eq!(
-        parse_full(
-            &("<b>".repeat(DEFAULT_DEPTH_LIMIT)
-                + " <s><i>test</i></s> "
-                + &*"</b>".repeat(DEFAULT_DEPTH_LIMIT))
-        ),
-        tok
-    );
-}
-
-#[test]
-fn parse_mention() {
-    assert_eq!(
-        parse_full("@tag"),
-        Token::Mention {
-            mention_type: MentionType::User,
-            name: "tag".into(),
-            host: None,
-        }
-    );
-
-    assert_eq!(
-        parse_full("email@notactuallyamenmtion.org"),
-        Token::PlainText("email@notactuallyamenmtion.org".into())
-    );
-
-    assert_eq!(
-        parse_full("hgsjlkdsa @tag fgahjsdkd"),
-        Token::Sequence(vec![
-            Token::PlainText("hgsjlkdsa ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: None,
-            },
-            Token::PlainText(" fgahjsdkd".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("hgsjlkdsa @tag@ fgahjsdkd"),
-        Token::Sequence(vec![
-            Token::PlainText("hgsjlkdsa ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: None,
-            },
-            Token::PlainText("@ fgahjsdkd".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("aaaa @tag@domain bbbbb"),
-        Token::Sequence(vec![
-            Token::PlainText("aaaa ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: Some("domain".into()),
-            },
-            Token::PlainText(" bbbbb".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("test @tag@domain, test"),
-        Token::Sequence(vec![
-            Token::PlainText("test ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: Some("domain".into()),
-            },
-            Token::PlainText(", test".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("test @tag@domain.gay. test"),
-        Token::Sequence(vec![
-            Token::PlainText("test ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: Some("domain.gay".into()),
-            },
-            Token::PlainText(". test".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("test @tag@domain? test"),
-        Token::Sequence(vec![
-            Token::PlainText("test ".into()),
-            Token::Mention {
-                mention_type: MentionType::User,
-                name: "tag".into(),
-                host: Some("domain".into()),
-            },
-            Token::PlainText("? test".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("test !tag@domain.com test"),
-        Token::Sequence(vec![
-            Token::PlainText("test ".into()),
-            Token::Mention {
-                mention_type: MentionType::Community,
-                name: "tag".into(),
-                host: Some("domain.com".into()),
-            },
-            Token::PlainText(" test".into()),
-        ])
-    );
-
-    assert_eq!(
-        parse_full("@tag:domain.com"),
-        Token::Mention {
-            mention_type: MentionType::MatrixUser,
-            name: "tag".into(),
-            host: Some("domain.com".into())
-        },
-    );
-}
-
-#[test]
-fn parse_shortcodes() {
-    assert_eq!(
-        parse_full(":bottom:"),
-        Token::ShortcodeEmoji {
-            shortcode: "bottom".into(),
-            host: None,
-        }
-    );
-
-    assert_eq!(
-        parse_full(":bottom::blobfox:"),
-        Token::Sequence(vec![
-            Token::ShortcodeEmoji {
-                shortcode: "bottom".into(),
-                host: None,
-            },
-            Token::ShortcodeEmoji {
-                shortcode: "blobfox".into(),
-                host: None,
-            },
-        ])
-    );
-
-    assert_eq!(
-        parse_full(":bottom@magnetar.social:"),
-        Token::ShortcodeEmoji {
-            shortcode: "bottom".into(),
-            host: Some("magnetar.social".into()),
-        }
-    );
-
-    assert_eq!(
-        parse_full(":bottom:blobfox"),
-        Token::PlainText(":bottom:blobfox".into())
-    );
-
-    assert_eq!(
-        parse_full("bottom:blobfox:"),
-        Token::PlainText("bottom:blobfox:".into())
-    );
-}
-
-#[test]
-fn parse_emoji() {
-    assert_eq!(
-        parse_full("🥺💜❤️🦊"),
-        Token::Sequence(
-            vec!["🥺", "💜", "❤️", "🦊"]
-                .into_iter()
-                .map(str::to_string)
-                .map(Token::UnicodeEmoji)
-                .collect::<Vec<_>>()
-        )
-    );
-
-    // Trans flag, ZWJ
-    assert_eq!(
-        parse_full("\u{1f3f3}\u{0fe0f}\u{0200d}\u{026a7}\u{0fe0f}"),
-        Token::UnicodeEmoji("\u{1f3f3}\u{0fe0f}\u{0200d}\u{026a7}\u{0fe0f}".into())
-    );
-
-    assert_eq!(
-        parse_full("\u{0200d}\u{1f3f3}\u{0fe0f}"),
-        Token::Sequence(vec![
-            Token::PlainText("\u{0200d}".into()),             // ZWJ
-            Token::UnicodeEmoji("\u{1f3f3}\u{0fe0f}".into()), // White flag
-        ])
-    );
-
-    // Trans flag, ZWNJ
-    assert_eq!(
-        parse_full("\u{1f3f3}\u{0fe0f}\u{0200c}\u{026a7}\u{0fe0f}"),
-        Token::Sequence(vec![
-            Token::UnicodeEmoji("\u{1f3f3}\u{0fe0f}".into()), // White flag
-            Token::PlainText("\u{0200c}".into()),             // ZWNJ
-            Token::UnicodeEmoji("\u{026a7}\u{0fe0f}".into()), // Trans symbol
-        ])
-    );
-
-    assert_eq!(
-        parse_full("\u{1f3f3}\u{0fe0f}\u{0200d}\u{0200d}\u{0200d}"),
-        Token::Sequence(vec![
-            Token::UnicodeEmoji("\u{1f3f3}\u{0fe0f}".into()), // White flag
-            Token::PlainText("\u{0200d}\u{0200d}\u{0200d}".into()), // ZWJ
-        ])
-    );
-}
-
-#[test]
-fn xml_serialization() {
-    assert_eq!(
-        &to_xml_string(&parse_full("***nyaaa***")).unwrap(),
-        r#"<mmm><b><i>nyaaa</i></b></mmm>"#
-    );
-
-    assert_eq!(
-        &to_xml_string(&parse_full(
-            "@natty $[spin.speed=0.5s 🥺]:cat_attack: <plain>test</plain>"
-        ))
-            .unwrap(),
-        r#"<mmm><mention name="natty" type="user"/> <fn name="spin" arg-speed="0.5s"><ue>🥺</ue></fn><ee>cat_attack</ee> test</mmm>"#
-    );
-
-    assert_eq!(
-        &to_xml_string(&parse_full(
-            r#"
-```js
-var x = undefined;
-```         "#
-        ))
-            .unwrap(),
-        "<mmm><code lang=\"js\">var x = undefined;</code></mmm>"
-    );
-}

magnetar_mmm_parser/src/types.rs

@@ -1,120 +0,0 @@
-use unicode_segmentation::{Graphemes, UnicodeSegmentation};
-
-#[derive(Debug, Copy, Clone)]
-pub(crate) struct ParseSpan<'a> {
-    pub(crate) source: &'a str,
-    pub(crate) offset: usize,
-    pub(crate) length: usize,
-}
-
-impl ParseSpan<'_> {
-    pub(crate) fn concat(self, other: Self) -> Option<Self> {
-        if self.source != other.source {
-            panic!("Attempted to concat slices from different strings");
-        }
-
-        if self.offset + self.length != other.offset {
-            return None;
-        }
-
-        Some(ParseSpan {
-            source: self.source,
-            offset: self.offset,
-            length: self.length + other.length,
-        })
-    }
-
-    pub(crate) fn spanned_source(&self) -> &str {
-        &self.source[self.offset..self.offset + self.length]
-    }
-}
-
-
-pub(crate) struct TokStream<'a>(ParseSpan<'a>, Graphemes<'a>);
-
-impl<'a> From<&'a str> for TokStream<'a> {
-    fn from(source: &'a str) -> Self {
-        TokStream(
-            ParseSpan {
-                source,
-                length: source.len(),
-                offset: 0,
-            },
-            source.graphemes(true),
-        )
-    }
-}
-
-pub(crate) trait Input<'a> {
-    fn next(&mut self) -> Option<&'a str>;
-    fn view(&self) -> &'a str;
-}
-
-impl<'a> Input<'a> for TokStream<'a> {
-    #[inline]
-    fn next(&mut self) -> Option<&'a str> {
-        if let Some(p) = self.1.next() {
-            let length = p.len();
-            self.0.offset += length;
-            self.0.length -= length;
-            return Some(p);
-        }
-
-        None
-    }
-
-    #[inline]
-    fn view(&self) -> &'a str {
-        &self.0.source[self.0.offset..self.0.offset + self.0.length]
-    }
-}
-
-#[derive(Debug, Copy, Clone)]
-pub(crate) struct Lex<'a> {
-    pub(crate) token: &'a str,
-    pub(crate) span: ParseSpan<'a>,
-}
-
-pub(crate) type OutTok<'a> = Lex<'a>;
-
-pub(crate) const MAX_DEPTH: usize = 24;
-
-#[derive(Debug, Default, Clone, Copy)]
-pub(crate) struct State {
-    pub(crate) depth: usize,
-}
-
-pub(crate) enum Effect<'a> {
-    Output(OutTok<'a>)
-}
-
-
-#[must_use]
-pub(crate) struct ParserRet {
-    _private: (),
-}
-
-pub(crate) trait ParserCont {
-    fn continue_with(self, to: impl Parser) -> ParserRet;
-    fn continue_with2(self, to: (impl Parser, impl Parser)) -> ParserRet;
-}
-
-pub(crate) trait Parser {
-    fn take<'a>(
-        &mut self,
-        state: &mut State,
-        input: &mut impl Input<'a>,
-        handler: &'_ mut impl FnMut(Effect<'a>),
-        visitor: impl ParserCont,
-    ) -> ParserRet;
-}
-
-impl<I, F, V> Parser for fn(&mut State, &mut I, &'_ mut F, V) -> ParserRet {
-    fn take<'a>(&mut self,
-                state: &mut State,
-                input: &mut impl Input<'a>,
-                handler: &'_ mut impl FnMut(Effect<'a>),
-                visitor: impl ParserCont) -> ParserRet {
-        self(state, input, handler, visitor)
-    }
-}

magnetar_mmm_parser/src/xml_write.rs

@@ -1,155 +0,0 @@
-use std::io::{Cursor, Write};
-
-use crate::output_types::Token;
-use quick_xml::events::{BytesText, Event};
-
-impl Token {
-    fn write<T: Write>(&self, writer: &mut quick_xml::Writer<T>) -> quick_xml::Result<()> {
-        match self {
-            Token::PlainText(plain) => {
-                writer.write_event(Event::Text(BytesText::new(plain.as_str())))?;
-            }
-            Token::Sequence(sequence) => {
-                sequence.iter().try_for_each(|item| item.write(writer))?;
-            }
-            Token::Quote(inner) => {
-                writer
-                    .create_element("quote")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Small(inner) => {
-                writer
-                    .create_element("small")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Bold(inner) => {
-                writer
-                    .create_element("b")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Italic(inner) => {
-                writer
-                    .create_element("i")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Center(inner) => {
-                writer
-                    .create_element("center")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Strikethrough(inner) => {
-                writer
-                    .create_element("s")
-                    .write_inner_content(|w| inner.write(w))?;
-            }
-            Token::PlainTag(plain) => {
-                writer.write_event(Event::Text(BytesText::new(plain.as_str())))?;
-            }
-            Token::InlineCode(code) => {
-                writer
-                    .create_element("inline-code")
-                    .write_text_content(BytesText::new(code))?;
-            }
-            Token::InlineMath(math) => {
-                writer
-                    .create_element("inline-math")
-                    .write_text_content(BytesText::new(math))?;
-            }
-            Token::UrlRaw(url) => {
-                writer
-                    .create_element("a")
-                    .with_attribute(("href", url.as_str()))
-                    .write_text_content(BytesText::new(url))?;
-            }
-            Token::UrlNoEmbed(url) => {
-                writer
-                    .create_element("a")
-                    .with_attribute(("href", url.as_str()))
-                    .with_attribute(("embed", "false"))
-                    .write_text_content(BytesText::new(url))?;
-            }
-            Token::Link { label, href, embed } => {
-                writer
-                    .create_element("a")
-                    .with_attribute(("href", href.as_str()))
-                    .with_attribute(("embed", if *embed { "true" } else { "false" }))
-                    .write_inner_content(|w| label.write(w))?;
-            }
-            Token::BlockCode { inner, lang } => {
-                let mut ew = writer.create_element("code");
-
-                if let Some(language) = lang {
-                    ew = ew.with_attribute(("lang", language.as_str()));
-                }
-
-                ew.write_text_content(BytesText::new(inner))?;
-            }
-            Token::BlockMath(math) => {
-                writer
-                    .create_element("math")
-                    .write_text_content(BytesText::new(math))?;
-            }
-            Token::Function {
-                inner,
-                name,
-                params,
-            } => {
-                let mut ew = writer
-                    .create_element("fn")
-                    .with_attribute(("name", name.as_str()));
-
-                for (k, v) in params {
-                    ew = ew
-                        .with_attribute((format!("arg-{k}").as_str(), v.as_deref().unwrap_or("")));
-                }
-
-                ew.write_inner_content(|w| inner.write(w))?;
-            }
-            Token::Mention {
-                name,
-                host,
-                mention_type,
-            } => {
-                let mut ew = writer
-                    .create_element("mention")
-                    .with_attribute(("name", name.as_str()))
-                    .with_attribute(("type", mention_type.into()));
-
-                if let Some(host) = host {
-                    ew = ew.with_attribute(("host", host.as_str()));
-                }
-
-                ew.write_empty()?;
-            }
-            Token::UnicodeEmoji(text) => {
-                writer
-                    .create_element("ue")
-                    .write_text_content(BytesText::new(text))?;
-            }
-            Token::ShortcodeEmoji { shortcode, host } => {
-                let mut ew = writer.create_element("ee");
-
-                if let Some(host) = host {
-                    ew = ew.with_attribute(("host", host.as_str()));
-                }
-
-                ew.write_text_content(BytesText::new(shortcode))?;
-            }
-            Token::Hashtag(tag) => {
-                writer
-                    .create_element("hashtag")
-                    .write_text_content(BytesText::new(tag.as_str()))?;
-            }
-        }
-
-        Ok(())
-    }
-}
-
-pub fn to_xml_string(token: &Token) -> quick_xml::Result<String> {
-    let mut writer = quick_xml::Writer::new(Cursor::new(Vec::new()));
-    writer
-        .create_element("mmm")
-        .write_inner_content(|writer| token.write(writer))?;
-    Ok(String::from_utf8(writer.into_inner().into_inner())?)
-}

src/rpc_v1/mod.rs

@@ -23,7 +23,7 @@ struct RpcApGet {
 struct RpcApPost {
     user_id: String,
     url: String,
-    body: serde_json::Value,
+    body: String,
 }
 
 pub fn create_rpc_router() -> MagRpc {
@@ -43,9 +43,9 @@ pub fn create_rpc_router() -> MagRpc {
                     attachments: true,
                     with_context: true,
                 }
-                .fetch_single(&ctx, &id)
+                    .fetch_single(&ctx, &id)
-                .await?
+                    .await?
-                .ok_or(ObjectNotFound(id))?;
+                    .ok_or(ObjectNotFound(id))?;
 
                 Result::<_, ApiError>::Ok(note)
             },
@@ -90,7 +90,7 @@ pub fn create_rpc_router() -> MagRpc {
                     .create_signing_key(&key_id, SigningAlgorithm::RsaSha256)?;
                 let result = service
                     .ap_client
-                    .signed_post(signing_key, SigningAlgorithm::RsaSha256, None, &url, &body)
+                    .signed_post(signing_key, SigningAlgorithm::RsaSha256, None, &url, body.into_bytes())
                     .await?;
                 Result::<_, DeliveryError>::Ok(result)
             },

src/rpc_v1/proto.rs

@@ -1,21 +1,26 @@
 use crate::service::MagnetarService;
 use either::Either;
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
+use futures::{FutureExt, Stream, TryFutureExt, TryStreamExt};
+use futures_util::future::BoxFuture;
+use futures_util::stream::FuturesUnordered;
+use futures_util::{pin_mut, SinkExt, StreamExt};
 use miette::{miette, IntoDiagnostic};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::Debug;
+use std::future;
 use std::future::Future;
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
 use tokio::net::{TcpListener, UnixSocket};
 use tokio::select;
 use tokio::task::JoinSet;
+use tokio::time::Instant;
 use tracing::{debug, error, info, warn, Instrument};
 
 #[derive(Debug, Clone)]
@@ -57,7 +62,7 @@ pub struct RpcResult<T> {
 }
 
 impl<T: Serialize + Send + 'static, E: Serialize + Debug + Send + 'static> IntoRpcResponse
-    for Result<T, E>
+for Result<T, E>
 {
     fn into_rpc_response(self) -> Option<RpcResponse> {
         match self {
@@ -65,21 +70,21 @@ impl<T: Serialize + Send + 'static, E: Serialize + Debug + Send + 'static> IntoR
                 success: true,
                 data,
             })
-            .into_rpc_response(),
+                .into_rpc_response(),
             Err(data) => {
                 warn!("{:?}", data);
                 RpcMessage(RpcResult {
                     success: false,
                     data,
                 })
-                .into_rpc_response()
+                    .into_rpc_response()
             }
         }
     }
 }
 
 impl<L: IntoRpcResponse + Send + 'static, R: IntoRpcResponse + Send + 'static> IntoRpcResponse
-    for Either<L, R>
+for Either<L, R>
 {
     fn into_rpc_response(self) -> Option<RpcResponse> {
         match self {
@@ -97,14 +102,14 @@ where
         &self,
         context: Arc<MagnetarService>,
         message: RpcMessage<T>,
-    ) -> impl Future<Output = Option<RpcResponse>> + Send;
+    ) -> impl Future<Output=Option<RpcResponse>> + Send;
 }
 
 impl<T, F, Fut, RR> RpcHandler<T> for F
 where
     T: Send + 'static,
     F: Fn(Arc<MagnetarService>, RpcMessage<T>) -> Fut + Send + Sync + 'static,
-    Fut: Future<Output = RR> + Send,
+    Fut: Future<Output=RR> + Send,
     RR: IntoRpcResponse,
 {
     async fn process(
@@ -119,15 +124,15 @@ where
 type MessageRaw = Box<dyn Any + Send + 'static>;
 
 type MagRpcHandlerMapped = dyn Fn(
-        Arc<MagnetarService>,
+    Arc<MagnetarService>,
-        MessageRaw,
+    MessageRaw,
-    ) -> Pin<Box<dyn Future<Output = Option<RpcResponse>> + Send + 'static>>
+) -> Pin<Box<dyn Future<Output=Option<RpcResponse>> + Send + 'static>>
-    + Send
++ Send
-    + Sync
++ Sync
-    + 'static;
++ 'static;
 
 type MagRpcDecoderMapped =
-    dyn (Fn(&'_ [u8]) -> Result<MessageRaw, rmp_serde::decode::Error>) + Send + Sync + 'static;
+dyn (Fn(&'_ [u8]) -> Result<MessageRaw, rmp_serde::decode::Error>) + Send + Sync + 'static;
 
 pub struct MagRpc {
     listeners: HashMap<String, Arc<MagRpcHandlerMapped>>,
@@ -158,7 +163,7 @@ impl MagRpc {
                         .process(ctx, RpcMessage(*data.downcast().unwrap()))
                         .await
                 }
-                .boxed()
+                    .boxed()
             }),
         );
         self.payload_decoders.insert(
@@ -173,7 +178,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         addr: RpcSockAddr,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         match addr {
             RpcSockAddr::Ip(sock_addr) => {
@@ -187,7 +192,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         sock_addr: &SocketAddr,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         debug!("Binding RPC TCP socket to {}", sock_addr);
         let listener = TcpListener::bind(sock_addr).await.into_diagnostic()?;
@@ -205,6 +210,8 @@ impl MagRpc {
 
         loop {
             let (stream, remote_addr) = select!(
+                biased;
+                _ = &mut cancel => break,
                 Some(c) = connections.join_next() => {
                     debug!("RPC TCP connection closed: {:?}", c);
                     continue;
@@ -217,45 +224,20 @@ impl MagRpc {
 
                     conn.unwrap()
                 },
-                _ = &mut cancel => break
             );
 
             debug!("RPC TCP connection accepted: {:?}", remote_addr);
 
             let (cancel_send, cancel_recv) = tokio::sync::oneshot::channel::<()>();
-            let (read_half, mut write_half) = stream.into_split();
+            let (read_half, write_half) = stream.into_split();
-            let buf_read = BufReader::new(read_half);
+            let handler_fut = handle_process(
-            let context = context.clone();
+                rx_dec.stream_decode(BufReader::new(read_half), cancel_recv),
-            let rx_dec = rx_dec.clone();
+                BufWriter::new(write_half),
-            let fut = async move {
+                context.clone(),
-                let src = rx_dec
+            )
-                    .stream_decode(buf_read, cancel_recv)
+                .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
-                    .map_ok(process(context))
-                    .try_buffer_unordered(100)
-                    .boxed();
 
-                futures::pin_mut!(src);
+            connections.spawn(handler_fut);
-
-                while let Some(result) = src.try_next().await? {
-                    let Some((serial, RpcResponse(bytes))) = result else {
-                        continue;
-                    };
-
-                    write_half.write_u8(b'M').await.into_diagnostic()?;
-                    write_half.write_u64(serial).await.into_diagnostic()?;
-                    write_half
-                        .write_u32(bytes.len() as u32)
-                        .await
-                        .into_diagnostic()?;
-                    write_half.write_all(&bytes).await.into_diagnostic()?;
-                    write_half.flush().await.into_diagnostic()?;
-                }
-
-                Ok(remote_addr)
-            }
-            .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
-
-            connections.spawn(fut);
 
             cancellation_tokens.push(cancel_send);
         }
@@ -276,7 +258,7 @@ impl MagRpc {
         self,
         context: Arc<MagnetarService>,
         addr: &Path,
-        graceful_shutdown: Option<impl Future<Output = ()>>,
+        graceful_shutdown: Option<impl Future<Output=()>>,
     ) -> miette::Result<()> {
         let sock = UnixSocket::new_stream().into_diagnostic()?;
         debug!("Binding RPC Unix socket to {}", addr.display());
@@ -295,6 +277,8 @@ impl MagRpc {
 
         loop {
             let (stream, remote_addr) = select!(
+                biased;
+                _ = &mut cancel => break,
                 Some(c) = connections.join_next() => {
                     debug!("RPC Unix connection closed: {:?}", c);
                     continue;
@@ -306,46 +290,21 @@ impl MagRpc {
                     }
 
                     conn.unwrap()
-                },
+                }
-                _ = &mut cancel => break
             );
 
             debug!("RPC Unix connection accepted: {:?}", remote_addr);
 
             let (cancel_send, cancel_recv) = tokio::sync::oneshot::channel::<()>();
-            let (read_half, mut write_half) = stream.into_split();
+            let (read_half, write_half) = stream.into_split();
-            let buf_read = BufReader::new(read_half);
+            let handler_fut = handle_process(
-            let context = context.clone();
+                rx_dec.stream_decode(BufReader::with_capacity(64 * 1024, read_half), cancel_recv),
-            let rx_dec = rx_dec.clone();
+                BufWriter::new(write_half),
-            let fut = async move {
+                context.clone(),
-                let src = rx_dec
+            )
-                    .stream_decode(buf_read, cancel_recv)
+                .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
-                    .map_ok(process(context))
-                    .try_buffer_unordered(100)
-                    .boxed();
 
-                futures::pin_mut!(src);
+            connections.spawn(handler_fut);
-
-                while let Some(result) = src.try_next().await? {
-                    let Some((serial, RpcResponse(bytes))) = result else {
-                        continue;
-                    };
-
-                    write_half.write_u8(b'M').await.into_diagnostic()?;
-                    write_half.write_u64(serial).await.into_diagnostic()?;
-                    write_half
-                        .write_u32(bytes.len() as u32)
-                        .await
-                        .into_diagnostic()?;
-                    write_half.write_all(&bytes).await.into_diagnostic()?;
-                    write_half.flush().await.into_diagnostic()?;
-                }
-
-                miette::Result::<()>::Ok(())
-            }
-            .instrument(tracing::info_span!("RPC", remote_addr = ?remote_addr));
-
-            connections.spawn(fut.boxed());
 
             cancellation_tokens.push(cancel_send);
         }
@@ -363,16 +322,87 @@ impl MagRpc {
     }
 }
 
+async fn write_response(
+    mut buf_write: Pin<&mut BufWriter<impl AsyncWrite>>,
+    serial: u64,
+    result: Option<RpcResponse>,
+) -> miette::Result<()> {
+    let header = if result.is_some() { b'M' } else { b'F' };
+    buf_write.write_u8(header).await.into_diagnostic()?;
+    buf_write.write_u64(serial).await.into_diagnostic()?;
+
+    let Some(RpcResponse(bytes)) = result else {
+        return Ok(());
+    };
+
+    buf_write
+        .write_u32(bytes.len() as u32)
+        .await
+        .into_diagnostic()?;
+    buf_write.write_all(&bytes).await.into_diagnostic()?;
+    buf_write.flush().await.into_diagnostic()?;
+    Ok(())
+}
+
+async fn handle_process(
+    task_stream: impl Stream<Item=miette::Result<(u64, MessageRaw, Arc<MagRpcHandlerMapped>)>> + Send + 'static,
+    mut buf_write: BufWriter<impl AsyncWrite + Unpin>,
+    context: Arc<MagnetarService>,
+) -> miette::Result<()> {
+    let results = FuturesUnordered::new();
+    pin_mut!(results);
+
+    let (tx, rx) = futures::channel::mpsc::unbounded();
+    pin_mut!(rx);
+    let input_stream = tokio::spawn(
+        task_stream
+            .map_ok(process(context))
+            .boxed()
+            .forward(tx.sink_map_err(|e| miette!(e)))
+    );
+    pin_mut!(input_stream);
+
+    loop {
+        select!(
+            biased;
+            _ = &mut input_stream => {
+                break;
+            }
+            Some(task) = rx.next() => {
+                results.push(task);
+            }
+            Some(res) = results.next() => {
+                let (serial, result) = res?;
+                write_response(Pin::new(&mut buf_write), serial, result)
+                    .await?;
+            }
+            else => {
+                break;
+            }
+        );
+    }
+
+    Ok(())
+}
+
 fn process(
     context: Arc<MagnetarService>,
 ) -> impl Fn(
     (u64, MessageRaw, Arc<MagRpcHandlerMapped>),
-) -> Pin<
+) -> BoxFuture<'static, miette::Result<(u64, Option<RpcResponse>)>> {
-    Box<dyn Future<Output = miette::Result<Option<(u64, RpcResponse)>>> + Send + 'static>,
-> {
     move |(serial, payload, listener)| {
         let ctx = context.clone();
-        tokio::task::spawn(async move { Some((serial, listener(ctx, payload).await?)) })
+        tokio::task::spawn(async move {
+            let start = Instant::now();
+            let res = listener(ctx, payload).await;
+            let took = start.elapsed();
+            // TODO: Extract this into a config
+            if took.as_secs_f64() > 50.0 {
+                warn!("Handler took long: {} sec", took.as_secs_f64());
+            }
+
+            (serial, res)
+        }.instrument(tracing::info_span!("Request", ?serial)))
             .map_err(|e| miette!(e))
             .boxed()
     }
@@ -385,11 +415,11 @@ struct RpcCallDecoder {
 }
 
 impl RpcCallDecoder {
-    fn stream_decode<R: AsyncRead + AsyncReadExt + Unpin + Send + 'static>(
+    fn stream_decode(
         &self,
-        mut buf_read: BufReader<R>,
+        mut buf_read: impl AsyncBufRead + Send + Unpin + 'static,
         mut cancel: tokio::sync::oneshot::Receiver<()>,
-    ) -> impl Stream<Item = miette::Result<(u64, MessageRaw, Arc<MagRpcHandlerMapped>)>> + Send + 'static
+    ) -> impl Stream<Item=miette::Result<(u64, MessageRaw, Arc<MagRpcHandlerMapped>)>> + Send + 'static
     {
         let decoders = self.payload_decoders.clone();
         let listeners = self.listeners.clone();
@@ -450,8 +480,9 @@ impl RpcCallDecoder {
                 };
 
                 let Some((serial, name_slice, payload_slice)) = select! {
-                    read_result = read_fut => read_result,
+                    biased;
                     _ = &mut cancel => { break; }
+                    read_result = read_fut => read_result,
                 }? else {
                     break;
                 };

src/service/federation_client.rs

@@ -20,13 +20,13 @@ pub(super) fn new_federation_client_service(
             "magnetar/{} (https://{})",
             config.branding.version, config.networking.host
         ))
-        .into_diagnostic()?,
+            .into_diagnostic()?,
     )
-    .into_diagnostic()
+        .into_diagnostic()
 }
 
 pub(super) fn new_ap_client_service(
     federation_client: impl AsRef<FederationClient> + Send + Sync + 'static,
-) -> impl ApClientService<Error = ApClientError> {
+) -> impl ApClientService<Error=ApClientError> {
     ApClientServiceDefaultProvider::new(federation_client)
 }