fix(remote): align compression table ack semantics

Author: j5ik2o

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

@@ -137,6 +137,41 @@ fn inbound_manifest_advertisement_resolves_manifest_metadata() {
   ));
 }
 
+#[test]
+fn inbound_advertisement_accepts_peer_table_larger_than_local_max() {
+  let config = RemoteCompressionConfig::new().with_actor_ref_max(max(1));
+  let mut tables = TcpCompressionTables::new(config);
+  let frame = WireFrame::Control(ControlPdu::CompressionAdvertisement {
+    authority:  "remote@host:1".to_string(),
+    table_kind: CompressionTableKind::ActorRef,
+    generation: 9,
+    entries:    vec![
+      CompressionTableEntry::new(3, "/user/a".to_string()),
+      CompressionTableEntry::new(4, "/user/b".to_string()),
+    ],
+  });
+
+  let action = tables.handle_inbound_frame(frame, "local@host:2").unwrap();
+
+  assert!(matches!(
+    action,
+    InboundCompressionAction::Reply {
+      pdu: ControlPdu::CompressionAck {
+        authority,
+        table_kind: CompressionTableKind::ActorRef,
+        generation: 9,
+      },
+      authority: peer_authority,
+    } if authority == "local@host:2" && peer_authority.authority() == "remote@host:1"
+  ));
+  let action =
+    tables.handle_inbound_frame(envelope_frame(CompressedText::table_ref(4), None, None), "local@host:2").unwrap();
+  assert!(matches!(
+    action,
+    InboundCompressionAction::Forward(WireFrame::Envelope(pdu)) if pdu.recipient_path() == "/user/b"
+  ));
+}
+
 #[test]
 fn outbound_acknowledged_metadata_uses_table_refs() {
   let mut tables = TcpCompressionTables::new(RemoteCompressionConfig::new());

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

@@ -159,6 +159,8 @@ impl CompressionTable {
       if entry.advertised_generation == Some(generation) {
         entry.acknowledged_generation = Some(generation);
         applied = true;
+      } else {
+        entry.acknowledged_generation = None;
       }
     }
     if applied {
@@ -189,11 +191,6 @@ impl CompressionTable {
     if !self.is_enabled() {
       return Ok(false);
     }
-    if let Some(max) = self.max
-      && entries.len() > max.get()
-    {
-      return Err(WireError::InvalidFormat);
-    }
     if has_duplicate_entry_id(entries) {
       return Err(WireError::InvalidFormat);
     }

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

@@ -1,7 +1,7 @@
 use alloc::string::ToString;
 use core::num::NonZeroUsize;
 
-use super::CompressionTable;
+use super::{CompressionTable, CompressionTableEntryState};
 use crate::wire::{CompressionTableEntry, CompressionTableKind, WireError};
 
 fn max(value: usize) -> Option<NonZeroUsize> {
@@ -117,6 +117,27 @@ fn stale_ack_is_ignored() {
   assert_eq!(table.latest_pending_generation(), Some(generation));
 }
 
+#[test]
+fn ack_clears_entries_not_present_in_acknowledged_generation() {
+  let mut table = CompressionTable::new(max(2));
+  table.entries.push(CompressionTableEntryState {
+    id: 1,
+    literal: "/user/a".to_string(),
+    hit_count: 2,
+    advertised_generation: Some(1),
+    acknowledged_generation: Some(1),
+  });
+  table.entries.push(CompressionTableEntryState::new(2, "/user/b".to_string()));
+  table.entries[1].hit_count = 1;
+  table.entries[1].advertised_generation = Some(2);
+  table.latest_pending_generation = Some(2);
+
+  assert!(table.acknowledge(2));
+
+  assert_eq!(table.encode("/user/a").as_literal(), Some("/user/a"));
+  assert_eq!(table.encode("/user/b").as_table_ref(), Some(2));
+}
+
 #[test]
 fn inbound_advertisement_resolves_entry_ids() {
   let mut table = CompressionTable::new(max(4));
@@ -129,15 +150,15 @@ fn inbound_advertisement_resolves_entry_ids() {
 }
 
 #[test]
-fn inbound_advertisement_rejects_entries_over_configured_max() {
+fn inbound_advertisement_accepts_entries_over_local_advertisement_bound() {
   let mut table = CompressionTable::new(max(1));
   let entries =
     [CompressionTableEntry::new(9, "/user/a".to_string()), CompressionTableEntry::new(10, "/user/b".to_string())];
 
-  let err = table.apply_advertisement(7, &entries).unwrap_err();
+  assert_eq!(table.apply_advertisement(7, &entries), Ok(true));
 
-  assert_eq!(err, WireError::InvalidFormat);
-  assert!(table.is_empty());
+  assert_eq!(table.resolve(9), Some("/user/a"));
+  assert_eq!(table.resolve(10), Some("/user/b"));
 }
 
 #[test]