Voidly's Probe-to-Dataset Ingest Pipeline: Normalization, Quality Filtering, and TimescaleDB Indexing

Probe device (Tauri app, boringtun WireGuard)
       │
       │  protobuf batch over QUIC/443 (domain-fronted)
       │  ~200KB per upload, every 60 seconds
       ▼
Cloudflare Worker (ingest.voidly.ai)
       │  – Authenticate probe (Ed25519 device certificate)
       │  – Schema validation (required fields present)
       │  – Rate limiting (per probe: max 2 uploads/min)
       │  – Queue to Cloudflare Queue → forwards to Kafka bridge
       ▼
Kafka (3 brokers, topic: raw_measurements, 12 partitions)
       │  7-day retention, lz4 compression
       ├──▶ Rust normalization consumer (primary path)
       └──▶ Raw archival consumer (stores original bytes to S3)
       ▼
Rust normalization consumer
       │  – Deserialize protobuf
       │  – Normalize schema across probe versions
       │  – Enrich (GeoIP, ASN lookup, domain canonicalization)
       │  – Quality filter (drop invalid measurements)
       │  – Bulk COPY to TimescaleDB (batches of 1,000)
       ▼
TimescaleDB (measurements table)
       │
       ├──▶ Async classifier (sets interference_type / confidence_tier)
       ├──▶ Continuous aggregate refresh (hourly)
       └──▶ Nightly Parquet export → HuggingFace (CC BY 4.0)

// probe_upload.proto
syntax = "proto3";

message MeasurementBatch {
    string probe_id       = 1;   // SHA-256 of device certificate public key
    bytes  device_sig     = 2;   // Ed25519 signature over batch_hash
    bytes  batch_hash     = 3;   // SHA-256 of measurements bytes
    int64  batch_seq      = 4;   // monotonic, per-probe; allows dedup on server
    string probe_version  = 5;   // semver "0.9.2"

    repeated Measurement measurements = 6;
}

message Measurement {
    // Required in all versions
    int64  measured_at_unix_ms = 1;
    string target_url          = 2;
    string test_protocol       = 3;   // "dns"|"tcp"|"tls"|"http"|"https"
    int32  vantage_asn         = 4;
    string vantage_country     = 5;   // ISO 3166-1 alpha-2

    // Optional (added across versions — always check has_field)
    repeated string dns_addrs       = 10;
    string          dns_error_code  = 11;
    bool            tcp_connected   = 20;
    int32           tcp_connect_ms  = 21;
    bool            tls_ok          = 30;
    bool            tls_cert_valid  = 31;
    int32           tls_alert_code  = 32;
    int32           http_status     = 40;
    bytes           http_body_sha   = 41;
    bool            control_ok      = 50;
}

// ingest worker (TypeScript)
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    if (request.method !== 'POST' || !request.url.endsWith('/ingest')) {
      return new Response('Not found', { status: 404 });
    }

    const body = new Uint8Array(await request.arrayBuffer());
    const batch = MeasurementBatch.decode(body);

    // 1. Authenticate probe
    const cert = await loadDeviceCert(batch.probeId, env.KV);
    if (!cert) return new Response('Unknown probe', { status: 401 });

    const sigValid = await verifyEd25519(
      batch.deviceSig,
      batch.batchHash,
      cert.publicKeyBytes
    );
    if (!sigValid) return new Response('Invalid signature', { status: 401 });

    // 2. Check batch sequence (deduplication)
    const lastSeq = await env.KV.get('seq:' + batch.probeId, 'json') ?? -1;
    if (batch.batchSeq <= lastSeq) {
      return new Response('Duplicate batch', { status: 200 });  // 200, not error
    }
    await env.KV.put('seq:' + batch.probeId, String(batch.batchSeq));

    // 3. Validate schema (required fields)
    for (const m of batch.measurements) {
      if (!m.measuredAtUnixMs || !m.targetUrl || !m.testProtocol) {
        return new Response('Invalid measurement', { status: 400 });
      }
    }

    // 4. Rate limiting
    const rateKey = 'rate:' + batch.probeId + ':' + Math.floor(Date.now() / 60000);
    const calls = await env.KV.get(rateKey, 'json') ?? 0;
    if (calls >= 2) return new Response('Rate limited', { status: 429 });
    await env.KV.put(rateKey, String(calls + 1), { expirationTtl: 120 });

    // 5. Enqueue to Cloudflare Queue → Kafka bridge
    await env.MEASUREMENT_QUEUE.send({ batch: Array.from(body) });

    return new Response('OK', { status: 202 });
  }
}

struct NormalizedMeasurement {
    // Fields present in all versions (v0.1+)
    measured_at:       DateTime<Utc>,
    target_url:        String,
    target_domain:     String,    // extracted from URL
    test_protocol:     Protocol,
    vantage_asn:       i32,
    vantage_country:   CountryCode,
    probe_id:          String,
    probe_version:     Version,

    // Added in v0.3 — None if probe is older
    tcp_connect_ms:    Option<i16>,

    // Added in v0.5 — None if probe is older
    tls_alert_code:    Option<i16>,
    tls_cert_sni:      Option<String>,

    // Added in v0.7 — None if probe is older
    http_body_sha256:  Option<[u8; 32]>,

    // Enriched fields (always present, derived server-side)
    target_ip_geo:     Option<CountryCode>,  // GeoIP of first dns_addrs entry
    target_asn:        Option<i32>,
    control_reachable: bool,
}

fn normalize(raw: &proto::Measurement, probe_ver: &Version) -> NormalizedMeasurement {
    NormalizedMeasurement {
        measured_at: DateTime::from_timestamp_millis(raw.measured_at_unix_ms)
            .unwrap_or_else(|| Utc::now()),   // clock skew protection: cap at now()
        target_domain: extract_domain(&raw.target_url),
        tcp_connect_ms: if probe_ver >= &Version::parse("0.3.0").unwrap() {
            raw.tcp_connect_ms.map(|ms| ms.min(i16::MAX as i32) as i16)
        } else {
            None
        },
        // ...
    }
}

struct EnrichmentPipeline {
    geoip:      MaxmindReader,    // GeoLite2-City, updated weekly
    asn_db:     MaxmindReader,    // GeoLite2-ASN, updated weekly
    domain_psl: PublicSuffixList, // Mozilla PSL for registrable domain extraction
}

impl EnrichmentPipeline {
    fn enrich(&self, m: &mut NormalizedMeasurement) {
        // GeoIP + ASN lookup for first resolved IP
        if let Some(ip) = m.dns_addrs.first() {
            m.target_ip_geo = self.geoip.lookup(ip).map(|r| r.country_code);
            m.target_asn = self.asn_db.lookup(ip).map(|r| r.autonomous_system_number);
        }

        // Extract registrable domain (e.g., "www.bbc.co.uk" → "bbc.co.uk")
        m.target_registrable = self.domain_psl
            .registrable_domain(&m.target_domain)
            .map(|d| d.to_string());

        // Canonicalize country code (some probes send "UK" instead of "GB")
        m.vantage_country = normalize_country_code(m.vantage_country);
    }
}

fn quality_check(m: &NormalizedMeasurement) -> Option<&'static str> {
    // Control server unreachable: can't distinguish censorship from probe failure
    if !m.control_reachable {
        return Some("control_unreachable");
    }

    // Bogon target IP: measurement hit an RFC-1918 or loopback address
    if m.dns_addrs.iter().any(is_bogon) && !m.dns_addrs.iter().any(is_public) {
        return Some("bogon_resolution_only");
    }

    // Clock skew: measured_at more than 48h in the past (late upload accepted
    // but not used for real-time anomaly detection)
    if Utc::now() - m.measured_at > Duration::hours(48) {
        return Some("late_arrival_gt48h");
    }

    // Empty URL: happens with some older probe versions on parse error
    if m.target_url.is_empty() {
        return Some("empty_target_url");
    }

    // Probe blacklisted (operator revoked or probe flagged as compromised)
    if REVOKED_PROBES.contains(&m.probe_id.as_str()) {
        return Some("probe_revoked");
    }

    None  // passes quality check
}

# CloudWatch metric alarm: scale out when Kafka lag > 100K
resource "aws_cloudwatch_metric_alarm" "consumer_lag" {
  alarm_name          = "voidly-consumer-lag-high"
  metric_name         = "kafka_consumer_lag_sum"
  comparison_operator = "GreaterThanThreshold"
  threshold           = 100000
  evaluation_periods  = 2
  period              = 60
  alarm_actions       = [aws_autoscaling_policy.scale_out.arn]
}

# Typical spike handling:
# t+0m:  spike begins, lag starts rising
# t+2m:  alarm fires (2× 60s evaluation periods)
# t+4m:  new instance starts (AMI baked, ~90s launch)
# t+6m:  new instance consuming, lag begins clearing
# t+60m: spike ends, scale-in after 60min cooldown

-- Export yesterday's measurements to Parquet (via pg_parquet extension)
COPY (
    SELECT
        measurement_id, probe_id, vantage_country, vantage_asn,
        measured_at, target_url, target_domain, test_protocol,
        interference_type, interference_prob, confidence_tier,
        dns_resolved, tcp_connected, tls_handshake_ok, http_status_code,
        control_reachable, inference_dropped, probe_version
    FROM measurements
    WHERE measured_at >= CURRENT_DATE - INTERVAL '1 day'
      AND measured_at < CURRENT_DATE
    -- Exclude quality-filtered rows from public export
    -- (they are available in the raw S3 archive for researchers)
    -- Actually: include them with inference_dropped set, so users can filter
)
TO '/tmp/measurements_{date}/part-{n}.parquet'
WITH (FORMAT parquet, COMPRESSION 'zstd', ROW_GROUP_SIZE 100000);

-- Upload to HuggingFace via huggingface_hub Python API
# Partitioned as country_code={CC}/year_month={YYYY-MM}/part-{n}.parquet
# Each country×month parquet is ~5–50MB depending on probe density

Stage                                       p50     p99
──────────────────────────────────────────────────────────────────
Probe upload (QUIC, ~200KB batch)           1.2s    4.1s
Cloudflare Worker processing                8ms     45ms
Cloudflare Queue → Kafka bridge             12s     40s   (polling)
Kafka producer → TimescaleDB insert         18s     62s   (batch accumulation)
──────────────────────────────────────────────────────────────────
Probe measurement → database row            ~32s    ~110s (normal load)
Probe measurement → anomaly detection       ~38s    ~130s (+classifier run)
Probe measurement → public dataset          ~26hr   ~27hr (nightly export)