Apply rustfmt across runtime, channels, and tests

Pure formatting — `cargo fmt --all` reflowed long imports, async
function signatures with multi-line return types, and a few
expression-level chains that exceeded the 100-column wrap. No
behavioral changes; `cargo test --workspace` still 169/0/0 and
clippy stays clean with -D warnings.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: James Ray

crates/mx20022-channels/mx20022-channel-http/src/lib.rs

@@ -340,7 +340,10 @@ mod tests {
     use mx20022_channels::{InboundChannel, OutboundChannel, OutboundMessage};
     use tokio::sync::{mpsc, RwLock};
 
-    use super::{handle_post, HttpInboundChannel, HttpInboundConfig, HttpOutboundChannel, HttpOutboundConfig, InboundState};
+    use super::{
+        handle_post, HttpInboundChannel, HttpInboundConfig, HttpOutboundChannel,
+        HttpOutboundConfig, InboundState,
+    };
 
     #[tokio::test]
     async fn inbound_handler_enqueues_message() {
@@ -399,7 +402,10 @@ mod tests {
         assert_eq!(resp.status(), StatusCode::ACCEPTED);
 
         // Verify the message was queued.
-        let msg = rx.recv().await.expect("should receive pre-shutdown message");
+        let msg = rx
+            .recv()
+            .await
+            .expect("should receive pre-shutdown message");
         assert_eq!(msg.raw, "<Document/>");
 
         // Trigger graceful shutdown.

crates/mx20022-runtime/src/app.rs

@@ -790,7 +790,10 @@ impl ParticipantRegistry {
         #[cfg(test)]
         registry.register("correlation-key-setter", build_correlation_key_setter);
         #[cfg(test)]
-        registry.register("correlation-expectation-setter", build_correlation_expectation_setter);
+        registry.register(
+            "correlation-expectation-setter",
+            build_correlation_expectation_setter,
+        );
         registry
     }
 
@@ -1089,7 +1092,13 @@ impl Participant for SlowParticipant {
     fn name(&self) -> &str {
         "slow"
     }
-    async fn prepare(&self, _ctx: &mut Context) -> Result<mx20022_runtime_core::participant::Action, mx20022_runtime_core::participant::ParticipantError> {
+    async fn prepare(
+        &self,
+        _ctx: &mut Context,
+    ) -> Result<
+        mx20022_runtime_core::participant::Action,
+        mx20022_runtime_core::participant::ParticipantError,
+    > {
         tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
         Ok(mx20022_runtime_core::participant::Action::Prepared)
     }
@@ -1124,8 +1133,10 @@ impl Participant for CorrelationKeySetter {
     async fn prepare(
         &self,
         ctx: &mut Context,
-    ) -> Result<mx20022_runtime_core::participant::Action, mx20022_runtime_core::participant::ParticipantError>
-    {
+    ) -> Result<
+        mx20022_runtime_core::participant::Action,
+        mx20022_runtime_core::participant::ParticipantError,
+    > {
         ctx.put(
             "correlation.lookup_key",
             CorrelationLookupKey {
@@ -1177,8 +1188,10 @@ impl Participant for CorrelationExpectationSetter {
     async fn prepare(
         &self,
         ctx: &mut Context,
-    ) -> Result<mx20022_runtime_core::participant::Action, mx20022_runtime_core::participant::ParticipantError>
-    {
+    ) -> Result<
+        mx20022_runtime_core::participant::Action,
+        mx20022_runtime_core::participant::ParticipantError,
+    > {
         ctx.put(
             "correlation.expectation",
             mx20022_store::Expectation {
@@ -1605,7 +1618,13 @@ participants = [
             .expect("app should build");
 
         let err = app
-            .process("timeout-pipeline", "TX-TO-1", "http-in", "pacs.008", "<Document/>")
+            .process(
+                "timeout-pipeline",
+                "TX-TO-1",
+                "http-in",
+                "pacs.008",
+                "<Document/>",
+            )
             .await
             .expect_err("should timeout");
 
@@ -1648,8 +1667,7 @@ participants = [
 
     #[tokio::test]
     async fn correlation_match_response_invoked_for_committed_transaction() {
-        let config =
-            RuntimeConfig::parse(CORRELATION_MATCH_CONFIG).expect("config should parse");
+        let config = RuntimeConfig::parse(CORRELATION_MATCH_CONFIG).expect("config should parse");
         let app = RuntimeApp::from_config(&config)
             .await
             .expect("app should build");
@@ -1789,8 +1807,7 @@ participants = [
     #[tokio::test]
     async fn build_business_rule_validator_extracts_scheme() {
         for scheme in &["fednow", "sepa", "cbpr"] {
-            let toml =
-                builder_config!("business-rule-validator", format!("scheme = '{}'", scheme));
+            let toml = builder_config!("business-rule-validator", format!("scheme = '{}'", scheme));
             let config = RuntimeConfig::parse(&toml).expect("config should parse");
             let app = RuntimeApp::from_config(&config).await;
             assert!(
@@ -1828,10 +1845,7 @@ participants = [
         for scope in &["global", "message_type", "source_channel"] {
             let toml = builder_config!(
                 "rate-limiter",
-                format!(
-                    "rate_per_second = 50.0, burst = 100.0, scope = '{}'",
-                    scope
-                )
+                format!("rate_per_second = 50.0, burst = 100.0, scope = '{}'", scope)
             );
             let config = RuntimeConfig::parse(&toml).expect("config should parse");
             let app = RuntimeApp::from_config(&config).await;
@@ -1918,10 +1932,7 @@ participants = [
 
     #[tokio::test]
     async fn build_routing_engine_rejects_rule_without_destination() {
-        let toml = builder_config!(
-            "routing-engine",
-            "rules = [{message_type = 'pacs.008'}]"
-        );
+        let toml = builder_config!("routing-engine", "rules = [{message_type = 'pacs.008'}]");
         let config = RuntimeConfig::parse(&toml).expect("config should parse");
         let err = match RuntimeApp::from_config(&config).await {
             Ok(_) => panic!("rule without destination should fail"),
@@ -2116,8 +2127,7 @@ participants = [
 
     #[tokio::test]
     async fn unknown_participant_name_returns_error() {
-        let config =
-            RuntimeConfig::parse(UNKNOWN_PARTICIPANT_CONFIG).expect("config should parse");
+        let config = RuntimeConfig::parse(UNKNOWN_PARTICIPANT_CONFIG).expect("config should parse");
         match RuntimeApp::from_config(&config).await {
             Err(RuntimeBuildError::UnknownParticipant(name)) => {
                 assert_eq!(name, "nonexistent-participant");

crates/mx20022-runtime/src/engine.rs

@@ -681,10 +681,7 @@ auth_bearer_token = "secret"
     fn pick_free_port() -> u16 {
         let listener =
             std::net::TcpListener::bind("127.0.0.1:0").expect("must bind ephemeral port");
-        listener
-            .local_addr()
-            .expect("must read local addr")
-            .port()
+        listener.local_addr().expect("must read local addr").port()
     }
 
     #[cfg(feature = "channel-http")]
@@ -721,8 +718,8 @@ participants = [{{ name = "message-logger" }}]
     async fn run_pipelines_returns_ok_when_shutdown_signals_immediately() {
         use std::sync::Arc;
 
-        use crate::app::RuntimeApp;
         use super::run_pipelines;
+        use crate::app::RuntimeApp;
 
         let port = pick_free_port();
         let config = runnable_config(port);
@@ -733,10 +730,12 @@ participants = [{{ name = "message-logger" }}]
         );
         // Already-resolved future = signal the engine to drain on first poll.
         let shutdown = async {};
-        let result =
-            tokio::time::timeout(std::time::Duration::from_secs(5), run_pipelines(app, config, shutdown))
-                .await
-                .expect("engine should exit before timeout");
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            run_pipelines(app, config, shutdown),
+        )
+        .await
+        .expect("engine should exit before timeout");
         assert!(result.is_ok(), "engine returned error: {:?}", result.err());
     }
 
@@ -745,8 +744,8 @@ participants = [{{ name = "message-logger" }}]
     async fn run_pipelines_returns_no_supported_pipelines_when_no_inbound_starts() {
         use std::sync::Arc;
 
-        use crate::app::RuntimeApp;
         use super::{run_pipelines, EngineError};
+        use crate::app::RuntimeApp;
 
         let port = pick_free_port();
         let mut config = runnable_config(port);

crates/mx20022-runtime/src/main.rs

@@ -21,7 +21,7 @@ use mx20022_runtime::engine;
 use sha2::{Digest, Sha256};
 use tokio::sync::RwLock;
 use tracing_subscriber::reload::Handle;
-use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt};
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
 #[tokio::main]
 async fn main() {
@@ -220,7 +220,7 @@ async fn run(
 async fn shutdown_signal() {
     #[cfg(unix)]
     {
-        use tokio::signal::unix::{SignalKind, signal};
+        use tokio::signal::unix::{signal, SignalKind};
         let mut sigterm = match signal(SignalKind::terminate()) {
             Ok(stream) => stream,
             Err(error) => {

crates/mx20022-runtime/tests/e2e_pipeline.rs

@@ -159,7 +159,13 @@ async fn e2e_process_pacs008_commits_and_persists_context() {
         .expect("app should build");
 
     let report = app
-        .process("e2e", "TX-E2E-1", "http-in", "pacs.008", pacs008_xml("MSG-E2E-1"))
+        .process(
+            "e2e",
+            "TX-E2E-1",
+            "http-in",
+            "pacs.008",
+            pacs008_xml("MSG-E2E-1"),
+        )
         .await
         .expect("process should succeed");
 
@@ -218,7 +224,12 @@ async fn e2e_multi_participant_commits_with_context_entries() {
     );
     // Neither participant should have errored
     for pr in &report.participant_results {
-        assert!(pr.error.is_none(), "{} errored: {:?}", pr.participant, pr.error);
+        assert!(
+            pr.error.is_none(),
+            "{} errored: {:?}",
+            pr.participant,
+            pr.error
+        );
     }
 
     let store = app.store_handle();
@@ -377,7 +388,13 @@ async fn e2e_unknown_pipeline_returns_error() {
         .expect("app should build");
 
     let err = app
-        .process("nonexistent", "TX-ERR-1", "http-in", "pacs.008", "<Document/>")
+        .process(
+            "nonexistent",
+            "TX-ERR-1",
+            "http-in",
+            "pacs.008",
+            "<Document/>",
+        )
         .await
         .expect_err("should fail for unknown pipeline");
 
@@ -400,10 +417,7 @@ async fn e2e_rejected_message_type_returns_error() {
         .expect_err("should fail for unsupported message type");
 
     let msg = err.to_string();
-    assert!(
-        msg.contains("not accepted"),
-        "unexpected error: {msg}"
-    );
+    assert!(msg.contains("not accepted"), "unexpected error: {msg}");
 }
 
 #[tokio::test]
@@ -414,10 +428,22 @@ async fn e2e_multi_pipeline_routing() {
         .expect("app should build");
 
     assert_eq!(app.pipeline_count().await, 2);
-    assert!(app.accepts_message_type("pacs008-pipeline", "pacs.008").await);
-    assert!(app.accepts_message_type("pacs002-pipeline", "pacs.002").await);
-    assert!(!app.accepts_message_type("pacs008-pipeline", "pacs.002").await);
-    assert!(!app.accepts_message_type("pacs002-pipeline", "pacs.008").await);
+    assert!(
+        app.accepts_message_type("pacs008-pipeline", "pacs.008")
+            .await
+    );
+    assert!(
+        app.accepts_message_type("pacs002-pipeline", "pacs.002")
+            .await
+    );
+    assert!(
+        !app.accepts_message_type("pacs008-pipeline", "pacs.002")
+            .await
+    );
+    assert!(
+        !app.accepts_message_type("pacs002-pipeline", "pacs.008")
+            .await
+    );
 
     // Process through pipeline 1
     let report1 = app