Reorganize TcpSubscriptionStealer 1 (#2213)

* New struct for managing port subscriptions and redirections

* Struct integrated with TcpConnectionStealer

* Docs

* Changelog entry

* .......

* Moved TcpListener ot IpTablesRedirector

* Extended comment when we get a stale connection for a canceled subscription

changelog.d/+tcp-stealer-reorganized.internal.md

@@ -0,0 +1 @@
+Reorganized `TcpSubscriptionStealer` code - extracted port subscriptions into a separate struct.

mirrord/agent/src/error.rs

@@ -5,7 +5,7 @@ use mirrord_protocol::{
         LayerConnect,
     },
     tcp::{DaemonTcp, LayerTcpSteal},
-    FileRequest, FileResponse, Port,
+    FileRequest, FileResponse,
 };
 use thiserror::Error;
 
@@ -81,9 +81,6 @@ pub(crate) enum AgentError {
     #[error("Bollard failed with `{0}`")]
     Bollard(#[from] bollard::errors::Error),
 
-    #[error("Connection received from unexepcted port `{0}`")]
-    UnexpectedConnection(Port),
-
     #[error("LayerTcpSteal sender failed with `{0}`")]
     SendLayerTcpSteal(#[from] tokio::sync::mpsc::error::SendError<LayerTcpSteal>),
 
@@ -105,9 +102,6 @@ pub(crate) enum AgentError {
     #[error("Ran out of connections, dropping new connection")]
     ConnectionLimitReached,
 
-    #[error("An internal invariant of the agent was violated, this should not happen.")]
-    AgentInvariantViolated,
-
     #[error(r#"Failed to set socket flag PACKET_IGNORE_OUTGOING, this might be due to kernel version before 4.20.
     Original error `{0}`"#)]
     PacketIgnoreOutgoing(#[source] std::io::Error),

mirrord/agent/src/steal.rs

@@ -10,14 +10,12 @@ use mirrord_protocol::{
     ConnectionId, Port, RequestId,
 };
 use tokio::{
-    net::TcpListener,
     select,
     sync::{mpsc::Sender, oneshot},
 };
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
-use self::ip_tables::SafeIpTables;
 use crate::{
     error::{AgentError, Result},
     steal::http::Response,
@@ -29,6 +27,7 @@ pub(super) mod connection;
 pub(super) mod http;
 pub(super) mod ip_tables;
 mod orig_dst;
+mod subscriptions;
 
 /// Commands from the agent that are passed down to the stealer worker, through [`TcpStealerApi`].
 ///