Harden Usenet threading and ingress deliveryHEADmain

1 files changed, 39 insertions, 7 deletions

diff --git a/ingress/src/main.rs b/ingress/src/main.rs
index 21da30f..39047ee 100644
--- a/ingress/src/main.rs
+++ b/ingress/src/main.rs
@@ -11,6 +11,7 @@ use tokio_socks::tcp::Socks5Stream;
 const MAX_REQUEST_BYTES: usize = 128 * 1024;
 const MAX_PAYLOAD_BYTES: usize = 96 * 1024;
 const SMTP_TIMEOUT: Duration = Duration::from_secs(60);
+const TOR_CONNECT_ATTEMPTS: u32 = 3;
 
 #[derive(Debug, Deserialize)]
 struct Envelope {
@@ -123,13 +124,29 @@ async fn deliver(entry_address: &str, payload: &[u8]) -> Result<()> {
     let (proxy_host, proxy_port) = proxy.rsplit_once(':').context("invalid YAMN_TOR_SOCKS")?;
     let proxy_port: u16 = proxy_port.parse().context("invalid YAMN_TOR_SOCKS port")?;
     let target = format!("{host}:25");
-    let stream = tokio::time::timeout(
-        SMTP_TIMEOUT,
-        Socks5Stream::connect((proxy_host, proxy_port), target.as_str()),
-    )
-    .await
-    .context("Tor SMTP connection timed out")?
-    .context("connect to YAMN SMTP entry through Tor")?;
+    let mut attempt = 1;
+    let stream = loop {
+        let connection = tokio::time::timeout(
+            SMTP_TIMEOUT,
+            Socks5Stream::connect((proxy_host, proxy_port), target.as_str()),
+        )
+        .await;
+        match connection {
+            Ok(Ok(stream)) => break stream,
+            Ok(Err(_)) if attempt < TOR_CONNECT_ATTEMPTS => {
+                eprintln!("Tor SMTP connection attempt {attempt} failed; retrying");
+            }
+            Err(_) if attempt < TOR_CONNECT_ATTEMPTS => {
+                eprintln!("Tor SMTP connection attempt {attempt} timed out; retrying");
+            }
+            Ok(Err(error)) => {
+                return Err(error).context("connect to YAMN SMTP entry through Tor");
+            }
+            Err(error) => return Err(error).context("Tor SMTP connection timed out"),
+        }
+        tokio::time::sleep(tor_retry_delay(attempt)).await;
+        attempt += 1;
+    };
     let mut reader = BufReader::new(stream);
     let greeting = smtp_read_response(&mut reader).await?;
     if greeting / 100 != 2 {
@@ -189,6 +206,10 @@ async fn deliver(entry_address: &str, payload: &[u8]) -> Result<()> {
     Ok(())
 }
 
+fn tor_retry_delay(failed_attempt: u32) -> Duration {
+    Duration::from_secs(5 * u64::from(failed_attempt))
+}
+
 async fn run() -> Result<()> {
     let disconnected = MixnetClientBuilder::new_with_default_storage(storage_paths()?)
         .await
@@ -243,3 +264,14 @@ async fn main() {
         std::process::exit(1);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn tor_retry_delay_uses_bounded_linear_backoff() {
+        assert_eq!(tor_retry_delay(1), Duration::from_secs(5));
+        assert_eq!(tor_retry_delay(2), Duration::from_secs(10));
+    }
+}