Add mobile_verifier_compare crate: PG ↔ Trino reward-query parity CLI

Cargo.toml

@@ -16,6 +16,7 @@ members = [
   "mobile_config_cli",
   "mobile_packet_verifier",
   "mobile_verifier",
+  "mobile_verifier_compare",
   "price",
   "price_tracker",
   "reward_index",

mobile_verifier/src/heartbeats/mod.rs

@@ -28,7 +28,7 @@ use uuid::Uuid;
 use self::last_location::{LastLocation, LocationCache};
 
 /// Minimum number of heartbeats required to give a reward to the hotspot.
-const MINIMUM_HEARTBEAT_COUNT: i64 = 12;
+pub const MINIMUM_HEARTBEAT_COUNT: i64 = 12;
 
 #[derive(Clone)]
 pub struct Heartbeat {

mobile_verifier/src/speedtests.rs

@@ -30,7 +30,7 @@ use std::{
 use task_manager::{ManagedTask, TaskManager};
 use tokio::sync::mpsc::Receiver;
 
-const SPEEDTEST_AVG_MAX_DATA_POINTS: usize = 6;
+pub const SPEEDTEST_AVG_MAX_DATA_POINTS: usize = 6;
 // The limit must be 300 megabits per second.
 // Values in proto are in bytes/sec format.
 // Convert 300 megabits per second to bytes per second.

mobile_verifier_compare/.gitignore

@@ -0,0 +1,4 @@
+# Auto-written by `mobile-verifier-compare seed` on first run, holds
+# docker-compose-defaults for local dev. Regenerated automatically — not
+# meant to be committed.
+pkg/compare-trino-settings.local.toml

mobile_verifier_compare/Cargo.toml

@@ -0,0 +1,40 @@
+[package]
+name = "mobile-verifier-compare"
+version = "0.1.0"
+description = "Validation CLI: diffs mobile_verifier's Postgres reward-input queries against equivalent Trino/Iceberg queries, with a local-stack seed bootstrapper."
+license.workspace = true
+edition.workspace = true
+authors.workspace = true
+
+[[bin]]
+name = "mobile-verifier-compare"
+path = "src/main.rs"
+
+[dependencies]
+# Production crate we call into so the Postgres side of every comparison is
+# *byte-identical* to what the reward pipeline runs (zero SQL drift).
+mobile-verifier = { path = "../mobile_verifier" }
+
+# Trino client (no longer optional anywhere — it's the whole point of this crate).
+trino-client = { path = "../trino_client" }
+trino-rust-client = { version = "0.9" }
+
+# Settings + PG pool plumbing (same shape mobile_verifier uses).
+db-store = { path = "../db_store" }
+custom-tracing = { path = "../custom_tracing" }
+
+# General workspace deps. Kept in lockstep with mobile_verifier's set so this
+# crate sees the same versions as the functions it calls into.
+anyhow = { workspace = true }
+chrono = { workspace = true }
+clap = { workspace = true }
+config = { workspace = true }
+futures = { workspace = true }
+helium-crypto = { workspace = true, features = ["sqlx-postgres"] }
+rand = { workspace = true }
+rust_decimal = { workspace = true }
+serde = { workspace = true }
+sqlx = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+uuid = { workspace = true }

mobile_verifier_compare/pkg/compare-trino-settings.toml

@@ -0,0 +1,61 @@
+# Settings file for `mobile-verifier compare-trino`.
+#
+# This is a *minimal* config that only carries what the diff CLI needs:
+#   - a Postgres URL for the mobile_verifier database
+#   - a Trino endpoint to query the corresponding Iceberg tables
+#
+# The full mobile_verifier daemon needs much more (buckets, signing keypair,
+# config client, …). Don't reuse this file for `mobile-verifier server`.
+#
+# Usage:
+#   cargo run -p mobile-verifier -- -c ./compare-trino-settings.toml \
+#       compare-trino --start 2026-05-20T00:00:00Z --end 2026-05-21T00:00:00Z all
+#
+# Any value can be overridden by env var, prefix `MV__`, double underscore as
+# the path separator. Examples:
+#   MV__DATABASE__URL=postgres://...
+#   MV__TRINO__AUTH__PASSWORD=... mobile-verifier -c ... compare-trino ...
+
+# Optional: tracing filter. Defaults to "mobile_verifier=info".
+# log = "mobile_verifier=debug,sqlx=warn"
+
+# ── Postgres (mobile_verifier database) ──────────────────────────────────────
+# This is the same `[database]` block used by the rest of mobile_verifier; if
+# you already have a working mobile_verifier settings.toml, copy that block
+# here verbatim.
+[database]
+url = "postgres://postgres:postgres@127.0.0.1:5432/mobile_verifier"
+max_connections = 10
+
+# Optional CA bundle for TLS to RDS, etc.
+# ca_path = "/etc/ssl/certs/rds-ca-bundle.pem"
+
+# ── Trino ────────────────────────────────────────────────────────────────────
+# The Trino coordinator that fronts the Iceberg catalog holding `poc.*` and
+# `rewards.*`. `catalog` should be the catalog name configured in Trino for
+# your Iceberg REST catalog (commonly "iceberg").
+[trino]
+host = "trino.example.com"
+port = 443
+user = "mobile-verifier-compare"
+secure = true
+catalog = "iceberg"
+# `schema` is optional — every query in this CLI is fully qualified (poc.*,
+# rewards.*), so the default schema doesn't actually matter.
+# schema = "default"
+
+# For local dev against a Trino with a self-signed cert:
+# insecure_skip_tls_verify = true
+
+# Pick exactly one auth block, or omit entirely for anonymous access.
+
+# --- Basic auth ---
+# [trino.auth]
+# type = "basic"
+# username = "alice"
+# password = "hunter2"
+
+# --- JWT bearer token ---
+# [trino.auth]
+# type = "jwt"
+# token = "eyJhbGciOi..."

mobile_verifier_compare/src/data_sessions.rs

@@ -0,0 +1,117 @@
+use crate::diff::{DiffTable, FloatEq, DEFAULT_F64_EPSILON};
+use crate::CommonArgs;
+use anyhow::{Context, Result};
+use mobile_verifier::data_session::aggregate_hotspot_data_sessions_to_dc;
+use serde::{Deserialize, Serialize};
+use sqlx::{Pool, Postgres};
+use std::collections::BTreeMap;
+use trino_client::{Client as TrinoClient, Statement};
+use trino_rust_client::Trino;
+
+#[derive(Debug, Clone, Default, PartialEq)]
+struct DcSummary {
+    rewardable_bytes: u64,
+}
+
+impl FloatEq for DcSummary {
+    fn float_eq(&self, other: &Self, _: f64) -> bool {
+        self == other
+    }
+}
+
+#[derive(Debug, Clone, Trino, Serialize, Deserialize)]
+struct TrinoDcRow {
+    hotspot_key: String,
+    rewardable_bytes: i64,
+}
+
+/// Compares `rewardable_bytes` aggregated per hotspot.
+///
+/// PG side reads `hotspot_data_transfer_sessions` rows whose `burn_timestamp`
+/// falls in the window — that's the reward pipeline's *input*. Trino side
+/// reads `rewards.data_transfer` rows whose `[start_period, end_period)`
+/// overlaps the window — that's the pipeline's *output*.
+///
+/// `rewardable_bytes` is the only column that is apples-to-apples between the
+/// two tables — both sides report it in raw bytes. PG's `num_dcs` (DC count,
+/// an input to the burn step) and Trino's `dc_transfer_reward` (the paid
+/// reward in bones, an output of burn × price) live in different units, so
+/// they are deliberately not diffed here.
+///
+/// Mismatches are expected on sessions that exist in PG but didn't produce a
+/// reward (e.g. sub-DC dust, or sessions whose hotspot was banned at reward
+/// time). PG totals should be ≥ Trino totals in a normal epoch.
+const TRINO_SQL: &str = r#"
+    SELECT
+        hotspot_key,
+        cast(sum(rewardable_bytes) AS bigint) AS rewardable_bytes
+    FROM rewards.data_transfer
+    WHERE start_period < :end AND end_period > :start
+    GROUP BY hotspot_key
+"#;
+
+pub async fn run(pg: &Pool<Postgres>, trino: &TrinoClient, args: &CommonArgs) -> Result<()> {
+    let epoch = args.epoch();
+
+    let pg_raw = aggregate_hotspot_data_sessions_to_dc(pg, &epoch)
+        .await
+        .context("running aggregate_hotspot_data_sessions_to_dc")?;
+    let pg_rows: BTreeMap<String, DcSummary> = pg_raw
+        .into_iter()
+        .map(|(k, v)| {
+            (
+                k.to_string(),
+                DcSummary {
+                    rewardable_bytes: v.rewardable_bytes,
+                },
+            )
+        })
+        .collect();
+
+    let stmt = Statement::new(TRINO_SQL)
+        .bind("start", epoch.start)
+        .bind("end", epoch.end)
+        .typed::<TrinoDcRow>();
+    let trino_raw: Vec<TrinoDcRow> = trino
+        .get_all(stmt)
+        .await
+        .context("running Trino rewards.data_transfer query")?;
+    let trino_rows: BTreeMap<String, DcSummary> = trino_raw
+        .into_iter()
+        .map(|r| {
+            (
+                r.hotspot_key,
+                DcSummary {
+                    rewardable_bytes: r.rewardable_bytes.max(0) as u64,
+                },
+            )
+        })
+        .collect();
+
+    let mut table = DiffTable::new(
+        "data-sessions — rewardable_bytes parity (PG input vs Trino reward output)",
+        "hotspot_key",
+        "rewardable_bytes",
+    )
+    .with_note(
+        "PG sums hotspot_data_transfer_sessions (pipeline input); Trino sums rewards.data_transfer (pipeline output). \
+         Only `rewardable_bytes` is comparable across these two tables — DC count and dc_transfer_reward are different units. \
+         Mismatches are expected on sessions that didn't produce a reward (sub-DC dust, banned hotspots, etc.).",
+    )
+    .with_options(args.show_all, args.limit);
+    table.fill(
+        pg_rows,
+        trino_rows,
+        DEFAULT_F64_EPSILON,
+        |k| k.clone(),
+        |v| v.rewardable_bytes.to_string(),
+        |p, t| {
+            format!(
+                "{:+}",
+                p.rewardable_bytes as i128 - t.rewardable_bytes as i128,
+            )
+        },
+    );
+    table.print();
+    Ok(())
+}