pd-edge: refactor: cleanup http abi impl

Author: fffonion

pd-edge/src/abi_impl/http/exchange.rs

@@ -194,7 +194,7 @@ mod tests {
         classify_downstream_http1_fast_body, downstream_http1_fast_path_eligible,
         downstream_http1_fast_path_expects_continue, outbound_http1_fast_path_eligible,
     };
-    use crate::abi_impl::http::HttpVersionPreference;
+    use crate::abi_impl::http::version::HttpVersionPreference;
 
     #[test]
     fn downstream_http1_fast_path_accepts_chunked_and_fixed_requests() {

pd-edge/src/abi_impl/http/fast_path.rs

@@ -33,33 +33,6 @@ struct CachedHeaderBatch {
 static HEADER_BATCH_CACHE: OnceLock<Mutex<HashMap<HeaderBatchCacheKey, CachedHeaderBatch>>> =
     OnceLock::new();
 
-pub(super) fn parse_header_name(name: impl AsRef<str>) -> Result<HeaderName, VmError> {
-    let name = name.as_ref();
-    HeaderName::from_bytes(name.as_bytes())
-        .map_err(|_| VmError::HostError(format!("invalid header name '{name}'")))
-}
-
-pub(super) fn parse_header(
-    name: impl AsRef<str>,
-    value: impl AsRef<str>,
-) -> Result<(HeaderName, HeaderValue), VmError> {
-    let name = name.as_ref();
-    let value = value.as_ref();
-    let header_name = HeaderName::from_bytes(name.as_bytes())
-        .map_err(|_| VmError::HostError(format!("invalid header name '{name}'")))?;
-    let header_value = HeaderValue::from_str(value)
-        .map_err(|_| VmError::HostError(format!("invalid header value '{value}'")))?;
-    Ok((header_name, header_value))
-}
-
-pub(super) fn request_path_with_query(path: &str, query: &str) -> String {
-    if query.is_empty() {
-        path.to_string()
-    } else {
-        format!("{path}?{query}")
-    }
-}
-
 fn header_batch_cache() -> &'static Mutex<HashMap<HeaderBatchCacheKey, CachedHeaderBatch>> {
     HEADER_BATCH_CACHE.get_or_init(|| Mutex::new(HashMap::new()))
 }
@@ -92,7 +65,7 @@ fn cached_header_batch_matches(source: &HeaderBatchCacheSource, headers: &Value)
     }
 }
 
-pub(super) fn lookup_cached_header_batch(headers: &Value) -> Option<HeaderMap> {
+fn lookup_cached_header_batch(headers: &Value) -> Option<HeaderMap> {
     let key = header_batch_cache_key(headers)?;
     let mut guard = header_batch_cache().lock();
     let cached = guard.get(&key)?;
@@ -103,7 +76,7 @@ pub(super) fn lookup_cached_header_batch(headers: &Value) -> Option<HeaderMap> {
     None
 }
 
-pub(super) fn store_cached_header_batch(headers: &Value, parsed: &HeaderMap) {
+fn store_cached_header_batch(headers: &Value, parsed: &HeaderMap) {
     let (Some(key), Some(source)) = (
         header_batch_cache_key(headers),
         header_batch_cache_source(headers),
@@ -132,6 +105,86 @@ pub(super) fn store_cached_header_batch(headers: &Value, parsed: &HeaderMap) {
     );
 }
 
+pub(super) fn parse_string_header_batch(
+    headers: Value,
+    batch_name: &'static str,
+) -> Result<HeaderMap, VmError> {
+    match headers {
+        Value::Null => Ok(HeaderMap::new()),
+        Value::Array(values) => {
+            if let Some(parsed) = lookup_cached_header_batch(&Value::Array(values.clone())) {
+                return Ok(parsed);
+            }
+            if values.len() % 2 != 0 {
+                return Err(VmError::HostError(format!(
+                    "{batch_name} arrays must contain alternating name/value string pairs",
+                )));
+            }
+            let mut parsed = HeaderMap::new();
+            for pair in values.chunks(2) {
+                let Value::String(name) = &pair[0] else {
+                    return Err(VmError::HostError(format!(
+                        "{batch_name} array keys must be strings",
+                    )));
+                };
+                let Value::String(value) = &pair[1] else {
+                    return Err(VmError::HostError(format!(
+                        "{batch_name} array values must be strings",
+                    )));
+                };
+                let (header_name, header_value) = parse_header(name.as_str(), value.as_str())?;
+                parsed.insert(header_name, header_value);
+            }
+            store_cached_header_batch(&Value::Array(values.clone()), &parsed);
+            Ok(parsed)
+        }
+        Value::Map(entries) => {
+            if let Some(parsed) = lookup_cached_header_batch(&Value::Map(entries.clone())) {
+                return Ok(parsed);
+            }
+            let mut parsed = HeaderMap::new();
+            for (key, value) in entries.as_ref() {
+                let Value::String(name) = key else {
+                    return Err(VmError::HostError(format!(
+                        "{batch_name} map keys must be strings",
+                    )));
+                };
+                let Value::String(value) = value else {
+                    return Err(VmError::HostError(format!(
+                        "{batch_name} map values must be strings",
+                    )));
+                };
+                let (header_name, header_value) = parse_header(name.as_str(), value.as_str())?;
+                parsed.insert(header_name, header_value);
+            }
+            store_cached_header_batch(&Value::Map(entries.clone()), &parsed);
+            Ok(parsed)
+        }
+        _ => Err(VmError::HostError(format!(
+            "{batch_name} must be null, an array of alternating strings, or a string map",
+        ))),
+    }
+}
+
+pub(super) fn parse_header_name(name: impl AsRef<str>) -> Result<HeaderName, VmError> {
+    let name = name.as_ref();
+    HeaderName::from_bytes(name.as_bytes())
+        .map_err(|_| VmError::HostError(format!("invalid header name '{name}'")))
+}
+
+pub(super) fn parse_header(
+    name: impl AsRef<str>,
+    value: impl AsRef<str>,
+) -> Result<(HeaderName, HeaderValue), VmError> {
+    let name = name.as_ref();
+    let value = value.as_ref();
+    let header_name = HeaderName::from_bytes(name.as_bytes())
+        .map_err(|_| VmError::HostError(format!("invalid header name '{name}'")))?;
+    let header_value = HeaderValue::from_str(value)
+        .map_err(|_| VmError::HostError(format!("invalid header value '{value}'")))?;
+    Ok((header_name, header_value))
+}
+
 pub(super) fn headers_to_value_map(headers: &HeaderMap) -> Value {
     let mut values = BTreeMap::<String, Vec<String>>::new();
     for (name, value) in headers {
@@ -154,6 +207,14 @@ pub(super) fn headers_to_value_map(headers: &HeaderMap) -> Value {
     )
 }
 
+pub(super) fn request_path_with_query(path: &str, query: &str) -> String {
+    if query.is_empty() {
+        path.to_string()
+    } else {
+        format!("{path}?{query}")
+    }
+}
+
 pub(super) fn query_to_value_map(query: &str) -> Value {
     let mut values = BTreeMap::<String, Vec<String>>::new();
     for (name, value) in url::form_urlencoded::parse(query.as_bytes()) {

pd-edge/src/abi_impl/http/helpers.rs

@@ -1,79 +1,13 @@
 #[cfg(feature = "http")]
-use self::helpers::{
-    headers_to_value_map, is_valid_request_path, lookup_cached_header_batch, parse_header,
-    parse_header_name, query_to_value_map, request_path_with_query, serialize_query_pairs,
-    store_cached_header_batch,
-};
-
-#[cfg(feature = "http")]
-mod exchange;
-mod fast_path;
+pub(crate) mod exchange;
+pub(crate) mod fast_path;
 mod helpers;
-mod outbound_http1;
+pub(crate) mod outbound_http1;
 #[cfg(feature = "http")]
 mod request;
 #[cfg(feature = "http")]
-mod response;
-mod state;
-mod version;
+pub(crate) mod response;
+pub(crate) mod state;
+pub(crate) mod version;
 
-#[cfg(feature = "http")]
-pub(crate) use exchange::prepare_default_upstream_request;
-pub(crate) use fast_path::{
-    DownstreamHttp1FastBodyKind, MAX_DOWNSTREAM_HTTP1_FAST_BODY_BYTES,
-    classify_downstream_http1_fast_body_lazy, downstream_http1_fast_path_eligible_lazy,
-    downstream_http1_fast_path_expects_continue_lazy, outbound_http1_fast_path_eligible,
-};
-pub(crate) use outbound_http1::new_shared_plain_http1_sender_pool;
-pub(crate) use outbound_http1::{OutboundHttp1ForwardBody, OutboundHttp1ForwardResponse};
-#[cfg(feature = "http")]
-pub(crate) use response::parse_response_header_batch;
-#[cfg(feature = "websocket")]
-pub(crate) use state::DownstreamWebSocketTunnelPlan;
-#[cfg(feature = "websocket")]
-pub(crate) use state::HttpOutboundRequestNode;
-#[cfg(feature = "tls")]
-pub(crate) use state::attach_outbound_exchange_tls_transport;
-pub(crate) use state::is_hop_by_hop_header;
-#[cfg(feature = "tls")]
-pub(crate) use state::upstream_response_available;
-pub(crate) use state::{
-    AttachedHttpTransport, DownstreamConnectTunnelPlan, DownstreamConnectTunnelTarget,
-    DownstreamConnectionMetadata, DownstreamHttpListenerGoal, DownstreamPostResponsePlan,
-    Http1DownstreamResolution, HttpPlaneRuntimeServicesConfig, HttpUpstreamScheme,
-    InlineDownstreamHttpResponse, LazyRequestId, PromotedDownstreamTransport, ProxyStreamRegistry,
-    ResolvedHttpGraphResponse, ResolvedNativeHttp1DownstreamResponse,
-    ResolvedNativeLocalHttp1DownstreamResponse, ResolvedSnapshotHttp1DownstreamResponse,
-    SharedRuntimeServices, allocate_tcp_stream_handle, allocate_udp_socket_handle,
-    append_outbound_exchange_body, append_outbound_exchange_body_bytes,
-    append_response_output_body_bytes, attach_outbound_exchange_tcp_transport,
-    build_downstream_http_request_context, build_downstream_http_request_context_from_components,
-    consume_request_body_all, default_upstream_exchange_handle, default_upstream_udp_socket_handle,
-    finish_downstream_response_stream, materialize_downstream_response_body_source,
-    new_shared_http_plane_runtime_services, outbound_exchange_exists,
-    outbound_exchange_response_available, outbound_exchange_response_eof,
-    outbound_exchange_tls_flow, read_downstream_response_trailers,
-    read_outbound_exchange_response_all, read_outbound_exchange_response_next_chunk,
-    read_outbound_exchange_response_next_line, read_outbound_exchange_response_trailers,
-    read_request_body_next_chunk, read_request_body_next_line, read_upstream_response_all,
-    read_upstream_response_next_chunk, read_upstream_response_next_line, request_body_eof,
-    resolve_committed_http_graph_response, resolve_http_graph_response,
-    resolve_http1_downstream_response, schedule_downstream_http_handoff,
-    start_downstream_response_stream, start_native_default_upstream_http_forward_response,
-    take_promoted_downstream_transport, tcp_stream_exists, udp_socket_exists,
-    upstream_response_eof, write_downstream_response_stream_bytes,
-};
 pub use state::{HttpRequestContext, LazyHttpHeaders, ProxyVmContext, SharedProxyVmContext};
-#[cfg(test)]
-pub(crate) use state::{RequestPortField, RequestStringField};
-#[cfg(any(feature = "http", test))]
-pub(crate) use state::{
-    allocate_outbound_exchange_handle, ensure_outbound_exchange_response_started,
-    ensure_upstream_response_started, read_request_body_all,
-};
-#[cfg(feature = "webrtc")]
-pub(crate) use state::{
-    allocate_webrtc_connection_handle, default_upstream_webrtc_connection_handle,
-    webrtc_connection_exists,
-};
-pub(crate) use version::HttpVersionPreference;

pd-edge/src/abi_impl/http/mod.rs

@@ -27,7 +27,7 @@ use tokio::{
 };
 use vm::VmError;
 
-use crate::abi_impl::http::LazyHttpHeaders;
+use crate::abi_impl::http::state::LazyHttpHeaders;
 use crate::abi_impl::transport::HTTP11_ALPN_PROTOCOL;
 #[cfg(feature = "tls")]
 use crate::abi_impl::transport::{