Feeds

Stream each feed's real-time events and download its hourly and daily Parquet exports.

Each feed comes from a Client method that returns a Feed<T>, where T is the feed's event type. A Feed owns a clone of the client, so it is cheap to create and does not borrow the Client. Every feed has a real-time stream and hourly Parquet exports that roll up into daily exports. Streaming and exports need separate scopes on the API key. The event types are listed under Event types.

pub fn proxies(&self) -> Feed<ProxyEvent>
pub fn anonymizers(&self) -> Feed<AnonymizerEvent>
pub fn torrents(&self) -> Feed<TorrentEvent>
pub fn ja4t(&self) -> Feed<Ja4tEvent>
pub fn helios_http(&self) -> Feed<HeliosHttpEvent>
pub fn helios_https(&self) -> Feed<HeliosHttpsEvent>
pub fn helios_dns(&self) -> Feed<HeliosDnsEvent>
pub fn helios_adb(&self) -> Feed<HeliosAdbEvent>

What each feed carries:

• proxies carries residential, mobile, and datacenter proxy sightings.
• anonymizers carries VPN and anonymizer IP ranges.
• torrents carries torrents seen on the BitTorrent DHT and trackers.
• ja4t carries JA4T TCP fingerprints.
• The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, tunnel destinations, and Android Debug Bridge commands. Despite its name, helios_dns does not carry DNS queries.

See Firehose, Feeds, and Helios for the data behind each feed.

Streams

stream

Opens the feed's real-time event stream.

pub async fn stream(&self) -> Result<EventStream<T>>

• Connection: Each call opens a new connection. The server keeps it open for up to 30 minutes and then closes it, which ends the stream with None.
• Errors: A non-success response returns an Error::Api, and a 429 without retry_after means the key's monthly request budget is spent or it already has too many open streams.
• Timeout: The client's timeout covers opening the stream but not reading it.
• Reconnecting: The API has no resume cursor, so events sent between connections are missed.

Call stream again to reconnect:

loop {
    let mut events = client.proxies().stream().await?;
    while let Some(event) = events.next().await {
        let event = event?;
        println!(
            "{} {}",
            event.ip.unwrap_or_default(),
            event.provider.unwrap_or_default()
        );
    }
}

The loop reconnects immediately after a clean close. Opening a connection already uses the SDK's bounded retries, including Retry-After handling. Any error that remains after those retries, or occurs while reading the stream, stops this consumer. A 429 without Retry-After can indicate an exhausted monthly request budget or too many open streams; resolve that condition before reconnecting.

Every feed streams the same way. Only the event type changes:

let mut events = client.proxies().stream().await?;
while let Some(event) = events.next().await {
    let event = event?;
    println!(
        "{:?} {:?} {:?} {:?} {:?}",
        event.ip, event.provider, event.r#type, event.country_code, event.asn
    );
}

Every field of a proxy, anonymizer, torrent, or JA4T event is an Option, because the API may omit any of them. Helios events require their core fields: timestamp, tunnel_id, domain, port, protocol, and meta on HTTP, HTTPS, and DNS events, and session, sequential_id, command, and hash on ADB events. An event that lacks one returns Error::Decode, which ends the stream like any other failure.

HeliosHttpsEvent::details is None when the sensor could not parse the handshake, so check it before use.

EventStream

EventStream<T> yields a feed's events from stream, or its exports from all_exports.

next returns the next item, or None once the stream has ended.

pub async fn next(&mut self) -> Option<Result<T>>

• Results: Blank keepalive lines are skipped.
• Errors: A failure, such as a dropped connection, an event that doesn't decode, or a failed page request, is returned once, and the stream then ends.
• Cleanup: Dropping the stream closes its connection or stops fetching pages.
• Traits: EventStream<T> implements the futures Stream trait with Item = Result<T>, so it works with combinators from futures or tokio-stream.

Exports

Each feed has hourly Parquet exports that roll up into daily exports.

• Hourly exports: An hourly export is available until its day's daily export is rolled up, shortly after 00:30 UTC the next day. After that, requests for the hour return a 400 Error::Api.
• Daily exports: While a daily export is being rolled up, requests for it return a 404 Error::Api with retry_after set.

Export IDs

An export ID is one of:

• "latest"
• a UTC day as "YYYY-MM-DD"
• a UTC hour as "YYYY-MM-DD/H", where a zero-padded hour is also accepted

"latest" is the feed's freshest export. Export::id values can be passed as they are. Any other string returns Error::InvalidInput before a request is sent.

The Rust SDK has no constant or helper functions for export IDs, so build the string yourself.

list_exports

Returns one page of the feed's exports, newest first.

pub async fn list_exports(&self, limit: Option<u32>, cursor: Option<&str>) -> Result<ExportPage>

let page = client.proxies().list_exports(Some(24), None).await?;
for export in &page.exports {
    println!("{} {} {}", export.id, export.kind, export.row_count);
}
println!("next cursor: {:?}", page.next_cursor);

• Input: limit sets the page size, from 1 to 500, and None uses the API default of 100.
• Paging: Pass a page's next_cursor as cursor to fetch the page after it, or call all_exports to read every export across pages.

all_exports

Returns a stream of every export, newest first, starting at cursor when it is set.

pub fn all_exports(&self, limit: Option<u32>, cursor: Option<&str>) -> ExportStream
where
    T: 'static,

pub type ExportStream = EventStream<Export>;

let mut exports = client.proxies().all_exports(None, None);
while let Some(export) = exports.next().await {
    let export = export?;
    println!("{} {} {}", export.id, export.kind, export.size_bytes);
}

• Paging: It is not async: pages of limit exports are fetched lazily as the stream is read, and dropping the stream stops fetching.
• Errors: An error is returned once and ends the stream, including an Error::Protocol when the API returns a cursor it already returned.

export_meta

Returns the checksum, size, row count, and Parquet schema of an export.

pub async fn export_meta(&self, id: &str) -> Result<ExportMeta>

let meta = client.proxies().export_meta("latest").await?;
println!("{} {} {} {}", meta.id, meta.rows, meta.size, meta.checksum);
for column in &meta.schema.fields {
    println!("{} {}", column.name, column.r#type);
}

download_url

Returns a presigned URL for an export's Parquet file, as a reqwest::Url.

pub async fn download_url(&self, id: &str) -> Result<Url>

let url = client.proxies().download_url("latest").await?;
println!("{:?}", url.host_str());

• Lifetime: The URL is valid for 24 hours.
• Security: The URL grants access without an API key, so treat it as a secret.
• Use: It is useful for tools that read Parquet over HTTPS directly.
• Errors: A 4xx or 5xx response returns an Error::Api. A successful response that isn't a redirect, or a redirect without a valid Location, returns Error::Protocol.

download

Opens an export's Parquet file for reading.

pub async fn download(&self, id: &str) -> Result<Download>

• Security: The API key is sent only to the Synthient API, never to the storage host that serves the file, and errors never include the presigned URL.
• Errors: A non-success response from either returns an Error::Api. Up to 10 redirects from the storage host are followed, and more return Error::Protocol. Requests to the storage host are not retried.
• Timeout: The client's timeout covers opening the download but not reading it.

Download

Download holds the contents of an export's Parquet file, read in chunks as they arrive.

• content_length returns the size of the file in bytes, if the storage host sent it.
• next returns the next chunk, or None once the file has been read. Each chunk is a bytes::Bytes, which dereferences to [u8].

pub fn content_length(&self) -> Option<u64>
pub async fn next(&mut self) -> Option<Result<Bytes>>

• Traits: Download implements the futures Stream trait with Item = Result<Bytes>.
• Cleanup: Dropping it closes the connection.

The SDK does not verify the file. Compare its SHA-256 with Export::checksum or ExportMeta::checksum. This program, adapted from the repository's download_export example, also uses the sha2 and hex crates and Tokio's fs and io-util features:

use sha2::{Digest, Sha256};
use synthient::{Client, ExportKind};
use tokio::fs::File;
use tokio::io::{AsyncWriteExt, BufWriter};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new(std::env::var("SYNTHIENT_API_KEY")?)?;
    let feed = client.proxies();

    let mut exports = feed.all_exports(None, None);
    let mut daily = None;
    while let Some(export) = exports.next().await {
        let export = export?;
        if export.kind == ExportKind::Daily {
            daily = Some(export);
            break;
        }
    }
    let daily = daily.ok_or("no daily export found")?;

    let mut download = feed.download(&daily.id).await?;
    if let Some(size) = download.content_length() {
        println!("downloading {} ({size} bytes)", daily.id);
    }
    let mut file = BufWriter::new(File::create("proxies.parquet").await?);
    let mut hash = Sha256::new();
    while let Some(chunk) = download.next().await {
        let chunk = chunk?;
        hash.update(&chunk);
        file.write_all(&chunk).await?;
    }
    file.flush().await?;

    if hex::encode(hash.finalize()) != daily.checksum {
        return Err("checksum mismatch".into());
    }
    Ok(())
}