Building a High-Concurrency OSINT Engine in Rust: How I Managed 35+ Async Streams Without Exhausting Sockets

Description

For the past few months, I have been building Reconx, an open-source CLI tool designed for mapping external infrastructure and gathering threat intelligence. If you have ever done penetration testing or bug bounty hunting, you know the standard workflow: run five different Go or Python tools, pipe the outputs together, and write bash scripts to parse the mess. I wanted a single, unified engine that could handle dozens of sources concurrently, track state changes, and output clean data. Here is a look under the hood at how I built the network and concurrency architecture in Rust.

The Problem: Socket Exhaustion in Async Rust

When I first started building Reconx, I made a classic mistake. I had over 35 OSINT collectors (querying Shodan, Censys, VirusTotal, etc.), and every time a collector fired, it instantiated a new HTTP client. When you run that at scale using tokio, you immediately run into ephemeral port exhaustion and get slammed by API rate limits. Target servers drop connections, and the OS runs out of file descriptors.

The Solution: Centralized Connection Pooling

To fix this, I completely refactored how Reconx handles outbound traffic. Instead of letting collectors manage their own network state, I built a centralized HTTP engine inside src/http.rs. I created a single, shared reqwest::Client wrapped in an Arc (Atomic Reference Count). Every single collector now routes its requests through this unified pool.

/// Build a centralized reqwest client with proxy rotation and custom timeouts.
/// Every collector shares this unified engine.
pub fn build_client ( timeout_secs : u64 , proxy_url : Option <& str > ) -> reqwest :: Result < reqwest :: Client > {
    let mut builder = reqwest :: Client :: builder ()
        .timeout ( Duration :: from_secs ( timeout_secs ))
        .user_agent ( "Mozilla/5.0 (compatible; Reconx/0.1; +https://github.com/redshadow912/Reconx)" );

    // Wire up HTTP/SOCKS5 proxy if the user passed the --proxy flag
    if let Some ( proxy ) = proxy_url {
        if ! proxy .is_empty () {
            builder = builder .proxy ( Proxy :: all ( proxy ) ? );
        }
    }
    builder .build ()
}

/// Execute an HTTP request with built-in rate-limiting and exponential backoff
/// Automatically handles HTTP 429 (Too Many Requests) and 5xx Server Errors.
pub async fn request_with_retry (
    client : & reqwest :: Client ,
    url : & str ,
    max_retries : u32 ,
    rate_limiter : Option <& ApiRateLimiter > ,
) -> Result < reqwest :: Response , reqwest :: Error > {
    let mut retries = 0 ;
    loop {
        // Enforce per-API quotas (e.g. max 5 requests/sec for Shodan)
        if let Some ( limiter ) = rate_limiter {
            limiter .until_ready () .await ;
        }
        let response = client .get ( url ) .send () .await ? ;
        let status = response .status ();

        // If successful, or if it's a hard client error (like 404), return immediately
        if status .is_success () || ( status .is_client_error () && status .as_u16 () != 429 ) {
            return Ok ( response );
        }

        // If rate limited (429) or server error (5xx), apply exponential backoff
        if retries >= max_retries {
            return Ok ( response );
        }

        // Wait 500ms -> 1s -> 2s -> 4s before retrying
        let backoff = Duration :: from_millis ( 500 * 2u64 .pow ( retries ));
        tokio :: time :: sleep ( backoff ) .await ;
        retries += 1 ;
    }
}

This change brought massive benefits:

• TCP Connection Reuse: reqwest handles connection pooling under the hood, meaning subsequent requests to the same API reuse the existing socket.
• Global Proxy Routing: By passing proxy configurations (HTTP/SOCKS5) into this single client, every single OSINT module instantly gained proxy support without needing to touch their individual code files.

Managing State and Asset Diffing

One of my biggest frustrations with existing tools is running a scan a week later and having to manually figure out what changed. To solve this, I built an in-memory diffing engine (src/analyzers/diff_engine.rs). It ingests the previous scan's state and compares it against the live run.

use std :: collections :: HashSet ;
use crate :: models :: Finding ;

pub struct DiffEngine ;

impl DiffEngine {
    /// Compare a live scan against the previous historical state
    /// to extract actionable intelligence (new assets, removed assets, new vulns).
    pub fn diff ( previous : & [ Finding ], current : & [ Finding ]) -> DiffResult {
        // 1. Transform arrays into HashSets for O(1) lookups
        let prev_subdomains : HashSet < String > = previous .iter ()
            .filter_map (| f | {
                if let Finding :: Subdomain ( s ) = f { Some ( s .subdomain .clone ()) } else { None }
            })
            .collect ();
        let curr_subdomains : HashSet < String > = current .iter ()
            .filter_map (| f | {
                if let Finding :: Subdomain ( s ) = f { Some ( s .subdomain .clone ()) } else { None }
            })
            .collect ();

        let prev_vulns : HashSet < String > = previous .iter ()
            .filter_map (| f | {
                if let Finding :: Vulnerability ( v ) = f {
                    Some ( format! ( "{}:{}" , v .host , v .cve_id .as_deref () .unwrap_or ( & v .vulnerability_type )))
                } else { None }
            })
            .collect ();
        let curr_vulns : HashSet < String > = current .iter ()
            .filter_map (| f | {
                if let Finding :: Vulnerability ( v ) = f {
                    Some ( format! ( "{}:{}" , v .host , v .cve_id .as_deref () .unwrap_or ( & v .vulnerability_type )))
                } else { None }
            })
            .collect ();

        // 2. Compute exact diffs instantly using set mathematics
        let new_subdomains : Vec < String > = curr_subdomains .difference ( & prev_subdomains ) .cloned () .collect ();
        let removed_subdomains : Vec < String > = prev_subdomains .difference ( & curr_subdomains ) .cloned () .collect ();
        let new_vulnerabilities : Vec < String > = curr_vulns .difference ( & prev_vulns ) .cloned () .collect ();

        let total_new = new_subdomains .len () + new_vulnerabilities .len ();
        let total_removed = removed_subdomains .len ();

        DiffResult {
            new_subdomains ,
            removed_subdomains ,
            new_vulnerabilities ,
            total_new ,
            total_removed ,
            // ... (other fields omitted for brevity)
        }
    }
}

This engine feeds directly into the takeover_detector.rs and risk_scorer.rs modules, meaning Reconx doesn't just give you a list of subdomains-it tells you exactly what is new and what is vulnerable right now.

Takeaways

Writing a highly concurrent network tool in Rust forces you to think deeply about resource management. Moving from ad-hoc HTTP requests to a centralized, reference-counted connection pool completely stabilized the tool under heavy load.

The project is entirely open-source, and you can check out the full architecture here:

Repository: https://github.com/redshadow912/Reconx

If you are a Rust developer, I would love for you to poke around the codebase. I am particularly interested in feedback on the async stream handling and error mapping.