fix(remote): address wire compression review feedback

Author: j5ik2o

modules/remote-adaptor-std/src/transport/tcp/base.rs

@@ -275,6 +275,19 @@ impl TcpRemoteTransport {
     self.default_address.as_ref().map(ToString::to_string).unwrap_or_default()
   }
 
+  fn local_authority_from_addresses(addresses: &[Address], bound_port: u16) -> String {
+    addresses
+      .first()
+      .map(|address| {
+        if address.port() == 0 {
+          Address::new(address.system(), address.host(), bound_port).to_string()
+        } else {
+          address.to_string()
+        }
+      })
+      .unwrap_or_default()
+  }
+
   fn peer_key_for_address(address: &Address) -> String {
     alloc::format!("{}:{}", address.host(), address.port())
   }
@@ -405,11 +418,12 @@ impl RemoteTransport for TcpRemoteTransport {
     if self.running {
       return Err(TransportError::AlreadyRunning);
     }
+    let configured_local_addresses = self.configured_local_addresses.clone();
     let bound_addr = self.server.start_with_remote_events(
       self.inbound_txs.clone(),
       self.remote_event_tx.clone(),
       self.monotonic_epoch,
-      self.local_authority(),
+      move |bound_port| Self::local_authority_from_addresses(&configured_local_addresses, bound_port),
     )?;
     self.apply_bound_port_to_advertised_addresses(bound_addr.port());
     self.running = true;

modules/remote-adaptor-std/src/transport/tcp/base_test.rs

@@ -47,3 +47,21 @@ fn outbound_lane_key_ignores_correlation_id_for_same_actor_pair() {
 
   assert_eq!(outbound_lane_key_for_envelope(&first), outbound_lane_key_for_envelope(&second));
 }
+
+#[test]
+fn local_authority_from_addresses_rewrites_ephemeral_port() {
+  let addresses = vec![Address::new("local-sys", "127.0.0.1", 0)];
+
+  let authority = TcpRemoteTransport::local_authority_from_addresses(&addresses, 2551);
+
+  assert_eq!(authority, "local-sys@127.0.0.1:2551");
+}
+
+#[test]
+fn local_authority_from_addresses_preserves_fixed_port() {
+  let addresses = vec![Address::new("local-sys", "127.0.0.1", 2552)];
+
+  let authority = TcpRemoteTransport::local_authority_from_addresses(&addresses, 2551);
+
+  assert_eq!(authority, "local-sys@127.0.0.1:2552");
+}

modules/remote-adaptor-std/src/transport/tcp/compression.rs

@@ -98,29 +98,31 @@ impl TcpCompressionTables {
     EnvelopePdu::new_with_metadata(
       recipient_path,
       sender_path,
-      (pdu.correlation_hi(), pdu.correlation_lo()),
+      pdu.correlation_hi(),
+      pdu.correlation_lo(),
       pdu.priority(),
-      pdu.serializer_id(),
+      EnvelopePayload::new(pdu.serializer_id(), None, pdu.payload().clone()),
       manifest,
-      pdu.payload().clone(),
     )
     .with_redelivery_sequence(pdu.redelivery_sequence())
   }
 
   fn resolve_inbound_envelope(&self, pdu: EnvelopePdu) -> Result<EnvelopePdu, WireError> {
-    let recipient_path = resolve_text(&self.inbound_actor_refs, pdu.recipient_path_metadata())?;
+    let recipient_path =
+      CompressedText::literal(resolve_text(&self.inbound_actor_refs, pdu.recipient_path_metadata())?);
     let sender_path =
       pdu.sender_path_metadata().map(|metadata| resolve_text(&self.inbound_actor_refs, metadata)).transpose()?;
     let manifest =
       pdu.manifest_metadata().map(|metadata| resolve_text(&self.inbound_manifests, metadata)).transpose()?;
     Ok(
-      EnvelopePdu::new(
+      EnvelopePdu::new_with_metadata(
         recipient_path,
-        sender_path,
+        sender_path.map(CompressedText::literal),
         pdu.correlation_hi(),
         pdu.correlation_lo(),
         pdu.priority(),
-        EnvelopePayload::new(pdu.serializer_id(), manifest, pdu.payload().clone()),
+        EnvelopePayload::new(pdu.serializer_id(), None, pdu.payload().clone()),
+        manifest.map(CompressedText::literal),
       )
       .with_redelivery_sequence(pdu.redelivery_sequence()),
     )

modules/remote-adaptor-std/src/transport/tcp/compression_test.rs

@@ -26,8 +26,16 @@ fn envelope_frame(
   manifest: Option<CompressedText>,
 ) -> WireFrame {
   WireFrame::Envelope(
-    EnvelopePdu::new_with_metadata(recipient_path, sender_path, (1, 2), 1, 7, manifest, Bytes::from_static(b"hello"))
-      .with_redelivery_sequence(None),
+    EnvelopePdu::new_with_metadata(
+      recipient_path,
+      sender_path,
+      1,
+      2,
+      1,
+      EnvelopePayload::new(7, None, Bytes::from_static(b"hello")),
+      manifest,
+    )
+    .with_redelivery_sequence(None),
   )
 }
 

modules/remote-adaptor-std/src/transport/tcp/server.rs

@@ -8,9 +8,7 @@ use std::{
   time::Instant,
 };
 
-use fraktor_remote_core_rs::{
-  config::RemoteCompressionConfig, extension::RemoteEvent, transport::TransportError, wire::CompressionTableKind,
-};
+use fraktor_remote_core_rs::{config::RemoteCompressionConfig, extension::RemoteEvent, transport::TransportError};
 use futures::{SinkExt as _, StreamExt as _};
 use tokio::{
   net::{TcpListener, TcpStream},
@@ -21,11 +19,9 @@ use tokio::{
 use tokio_util::codec::Framed;
 
 use super::{
+  WireFrame,
   client::inbound_lane_index,
-  compression::{
-    InboundCompressionAction, TcpCompressionTables, compression_advertisement_interval,
-    next_compression_advertisement_tick,
-  },
+  compression::{InboundCompressionAction, TcpCompressionTables},
   connection_loss_reporter::ConnectionLossReporter,
   frame_codec::WireFrameCodec,
   inbound_frame_event::InboundFrameEvent,
@@ -80,19 +76,22 @@ impl TcpServer {
     }
   }
 
-  pub(crate) fn start_with_remote_events(
+  pub(crate) fn start_with_remote_events<F>(
     &mut self,
     inbound_txs: Vec<UnboundedSender<InboundFrameEvent>>,
     remote_event_tx: Option<Sender<RemoteEvent>>,
     monotonic_epoch: Instant,
-    local_authority: String,
-  ) -> Result<SocketAddr, TransportError> {
+    local_authority_for_bound_port: F,
+  ) -> Result<SocketAddr, TransportError>
+  where
+    F: FnOnce(u16) -> String, {
     if self.accept_task.is_some() {
       return Err(TransportError::AlreadyRunning);
     }
     let handle = Handle::try_current().map_err(|_| TransportError::NotAvailable)?;
     let listener = StdTcpListener::bind(&self.bind_addr).map_err(|_| TransportError::SendFailed)?;
     let bound_addr = listener.local_addr().map_err(|_| TransportError::SendFailed)?;
+    let local_authority = local_authority_for_bound_port(bound_addr.port());
     listener.set_nonblocking(true).map_err(|_| TransportError::SendFailed)?;
     let listener = TcpListener::from_std(listener).map_err(|_| TransportError::SendFailed)?;
     let frame_codec = self.frame_codec;
@@ -164,69 +163,43 @@ async fn read_loop(
   let mut framed = Framed::new(stream, frame_codec);
   let mut authority = None;
   let mut compression_tables = TcpCompressionTables::new(compression_config);
-  let mut actor_ref_advertisement_interval = compression_advertisement_interval(
-    compression_config.actor_ref_max(),
-    compression_config.actor_ref_advertisement_interval(),
-  );
-  let mut manifest_advertisement_interval = compression_advertisement_interval(
-    compression_config.manifest_max(),
-    compression_config.manifest_advertisement_interval(),
-  );
   let exit_cause = loop {
-    tokio::select! {
-      next = framed.next() => match next {
-        | Some(Ok(decoded)) => {
-          let decoded = match compression_tables.handle_inbound_frame(decoded, &local_authority) {
-            | Ok(InboundCompressionAction::Forward(frame)) => frame,
-            | Ok(InboundCompressionAction::Reply(pdu)) => {
-              if let Err(err) = framed.send(crate::transport::tcp::WireFrame::Control(pdu)).await {
-                tracing::warn!(?err, peer = %peer, "tcp server compression ack write error");
-                break Some(TransportError::SendFailed);
-              }
-              continue;
-            },
-            | Ok(InboundCompressionAction::Consumed) => continue,
-            | Err(err) => {
-              tracing::warn!(?err, peer = %peer, "tcp server compression frame error");
+    match framed.next().await {
+      | Some(Ok(decoded)) => {
+        let decoded = match compression_tables.handle_inbound_frame(decoded, &local_authority) {
+          | Ok(InboundCompressionAction::Forward(frame)) => frame,
+          | Ok(InboundCompressionAction::Reply(pdu)) => {
+            if let Err(err) = framed.send(WireFrame::Control(pdu)).await {
+              tracing::warn!(?err, peer = %peer, "tcp server compression ack write error");
               break Some(TransportError::SendFailed);
-            },
-          };
-          if let Some(frame_authority) = authority_for_frame(&decoded) {
-            authority = Some(frame_authority);
-          }
-          let lane_index = inbound_lane_index(&peer, authority.as_ref(), &decoded, inbound_txs.len());
-          let inbound_tx =
-            inbound_txs.get(lane_index).expect("inbound_lane_index returns an index within the inbound_txs lane count");
-          if inbound_tx
-            .send(InboundFrameEvent { peer: peer.clone(), authority: authority.clone(), frame: decoded })
-            .is_err()
-          {
-            // Receiver dropped — the transport is shutting down.
-            break None;
-          }
-        },
-        | Some(Err(err)) => {
-          tracing::warn!(?err, peer = %peer, "tcp frame decode error");
-          break Some(TransportError::SendFailed);
-        },
-        | None => break Some(TransportError::ConnectionClosed),
-      },
-      _ = next_compression_advertisement_tick(&mut actor_ref_advertisement_interval) => {
-        if let Some(frame) = compression_tables.create_advertisement(CompressionTableKind::ActorRef, &local_authority)
-          && let Err(err) = framed.send(frame).await
-        {
-          tracing::warn!(?err, peer = %peer, "tcp server actor-ref compression advertisement write error");
-          break Some(TransportError::SendFailed);
+            }
+            continue;
+          },
+          | Ok(InboundCompressionAction::Consumed) => continue,
+          | Err(err) => {
+            tracing::warn!(?err, peer = %peer, "tcp server compression frame error");
+            break Some(TransportError::SendFailed);
+          },
+        };
+        if let Some(frame_authority) = authority_for_frame(&decoded) {
+          authority = Some(frame_authority);
         }
-      },
-      _ = next_compression_advertisement_tick(&mut manifest_advertisement_interval) => {
-        if let Some(frame) = compression_tables.create_advertisement(CompressionTableKind::Manifest, &local_authority)
-          && let Err(err) = framed.send(frame).await
+        let lane_index = inbound_lane_index(&peer, authority.as_ref(), &decoded, inbound_txs.len());
+        let inbound_tx =
+          inbound_txs.get(lane_index).expect("inbound_lane_index returns an index within the inbound_txs lane count");
+        if inbound_tx
+          .send(InboundFrameEvent { peer: peer.clone(), authority: authority.clone(), frame: decoded })
+          .is_err()
         {
-          tracing::warn!(?err, peer = %peer, "tcp server manifest compression advertisement write error");
-          break Some(TransportError::SendFailed);
+          // Receiver dropped — the transport is shutting down.
+          break None;
         }
       },
+      | Some(Err(err)) => {
+        tracing::warn!(?err, peer = %peer, "tcp frame decode error");
+        break Some(TransportError::SendFailed);
+      },
+      | None => break Some(TransportError::ConnectionClosed),
     }
   };
   if let (Some(cause), Some(authority), Some(sender)) = (exit_cause, authority, remote_event_tx) {

modules/remote-adaptor-std/src/transport/tcp_test.rs

@@ -161,7 +161,9 @@ fn make_test_server() -> TcpServer {
 
 fn start_test_server(server: &mut TcpServer, inbound_tx: UnboundedSender<InboundFrameEvent>) -> SocketAddr {
   server
-    .start_with_remote_events(vec![inbound_tx], None, Instant::now(), String::from("local@127.0.0.1:0"))
+    .start_with_remote_events(vec![inbound_tx], None, Instant::now(), |bound_port| {
+      format!("local@127.0.0.1:{bound_port}")
+    })
     .expect("server should bind to a system-assigned port")
 }
 
@@ -859,8 +861,11 @@ async fn tcp_server_consumes_compression_advertisement_and_replies_with_ack() {
     ack,
     WireFrame::Control(ControlPdu::CompressionAck { table_kind: CompressionTableKind::ActorRef, generation: 7, .. })
   ));
-  let inbound = tokio::time::timeout(Duration::from_millis(100), server_inbound_rx.recv()).await;
-  assert!(inbound.is_err(), "compression advertisement must not reach the inbound event loop");
+  tokio::task::yield_now().await;
+  assert!(
+    matches!(server_inbound_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+    "compression advertisement must not reach the inbound event loop"
+  );
   server.shutdown();
 }
 

modules/remote-core/src/wire/compression_table.rs

@@ -106,6 +106,12 @@ impl CompressionTable {
       entry.hit_count = entry.hit_count.saturating_add(1);
       return;
     }
+    let Some(max) = self.max else {
+      return;
+    };
+    if self.entries.len() >= max.get() {
+      return;
+    }
     let entry_id = self.next_entry_id;
     self.next_entry_id = self.next_entry_id.saturating_add(1);
     let mut entry = CompressionTableEntryState::new(entry_id, literal.to_string());

modules/remote-core/src/wire/compression_table_test.rs

@@ -20,6 +20,19 @@ fn observe_updates_hit_count_without_duplicate_entry_ids() {
   assert_eq!(table.entry_id("/user/a"), Some(1));
 }
 
+#[test]
+fn observe_stops_adding_entries_at_configured_max() {
+  let mut table = CompressionTable::new(max(1));
+
+  table.observe("/user/a");
+  table.observe("/user/b");
+  table.observe("/user/a");
+
+  assert_eq!(table.len(), 1);
+  assert_eq!(table.entry_id("/user/b"), None);
+  assert_eq!(table.hit_count("/user/a"), Some(2));
+}
+
 #[test]
 fn max_accessor_returns_configured_bound() {
   let table = CompressionTable::new(max(2));
@@ -48,7 +61,7 @@ fn disabled_table_does_not_track_hits_or_advertise() {
 
 #[test]
 fn advertisement_is_bounded_and_deterministic() {
-  let mut table = CompressionTable::new(max(2));
+  let mut table = CompressionTable::new(max(3));
   table.observe("/user/a");
   table.observe("/user/b");
   table.observe("/user/b");
@@ -58,9 +71,10 @@ fn advertisement_is_bounded_and_deterministic() {
   let advertisement = table.create_advertisement(CompressionTableKind::ActorRef).unwrap();
 
   assert_eq!(advertisement.generation(), 1);
-  assert_eq!(advertisement.entries().len(), 2);
+  assert_eq!(advertisement.entries().len(), 3);
   assert_eq!(advertisement.entries()[0].literal(), "/user/b");
   assert_eq!(advertisement.entries()[1].literal(), "/user/c");
+  assert_eq!(advertisement.entries()[2].literal(), "/user/a");
 }
 
 #[test]

modules/remote-core/src/wire/control_codec.rs

@@ -26,6 +26,7 @@ const SUBKIND_FLUSH_REQUEST: u8 = 0x04;
 const SUBKIND_FLUSH_ACK: u8 = 0x05;
 const SUBKIND_COMPRESSION_ADVERTISEMENT: u8 = 0x06;
 const SUBKIND_COMPRESSION_ACK: u8 = 0x07;
+const MIN_COMPRESSION_ENTRY_BYTES: usize = 4 + 4;
 
 /// Zero-sized codec for [`ControlPdu`].
 #[derive(Clone, Copy, Debug, Default)]
@@ -176,6 +177,9 @@ fn decode_compression_entries(buf: &mut Bytes) -> Result<Vec<CompressionTableEnt
     return Err(WireError::Truncated);
   }
   let entry_count = buf.get_u32() as usize;
+  if entry_count > buf.remaining() / MIN_COMPRESSION_ENTRY_BYTES {
+    return Err(WireError::Truncated);
+  }
   let mut entries = Vec::with_capacity(entry_count);
   for _ in 0..entry_count {
     if buf.remaining() < 4 {

modules/remote-core/src/wire/envelope_codec.rs

@@ -9,6 +9,7 @@ use crate::{
     compressed_text::{
       decode_compressed_text, decode_option_compressed_text, encode_compressed_text, encode_option_compressed_text,
     },
+    envelope_payload::EnvelopePayload,
     envelope_pdu::EnvelopePdu,
     frame_header::KIND_ENVELOPE,
     primitives::{begin_frame, decode_bytes, encode_bytes, patch_frame_length, read_frame_header},
@@ -60,16 +61,18 @@ impl Codec<EnvelopePdu> for EnvelopeCodec {
     let serializer_id = buf.get_u32();
     let manifest = decode_option_compressed_text(buf)?;
     let payload = decode_bytes(buf)?;
-    Ok(EnvelopePdu::new_with_metadata(
-      recipient_path,
-      sender_path,
-      (correlation_hi, correlation_lo),
-      priority,
-      serializer_id,
-      manifest,
-      payload,
-    ))
-    .map(|pdu| pdu.with_redelivery_sequence(redelivery_sequence))
+    Ok(
+      EnvelopePdu::new_with_metadata(
+        recipient_path,
+        sender_path,
+        correlation_hi,
+        correlation_lo,
+        priority,
+        EnvelopePayload::new(serializer_id, None, payload),
+        manifest,
+      )
+      .with_redelivery_sequence(redelivery_sequence),
+    )
   }
 }