feat(collector): package USD-M artifact verifier

.github/workflows/acr-publish.yml

@@ -281,15 +281,19 @@ jobs:
           docker cp "$container_id:/usr/local/bin/binance-lob-archiver" artifact/binance-lob-archiver
           docker cp "$container_id:/usr/local/bin/polymarket-raw-ops" artifact/polymarket-raw-ops
           docker cp "$container_id:/usr/local/bin/binance-usdm-reference-collector" artifact/binance-usdm-reference-collector
+          docker cp "$container_id:/usr/local/bin/binance-usdm-reference-artifact-verifier" artifact/binance-usdm-reference-artifact-verifier
           chmod 0755 artifact/binance-lob-archiver
           chmod 0755 artifact/polymarket-raw-ops
           chmod 0755 artifact/binance-usdm-reference-collector
+          chmod 0755 artifact/binance-usdm-reference-artifact-verifier
           (
             cd artifact
             sha256sum binance-lob-archiver | tee binance-lob-archiver.sha256
             sha256sum polymarket-raw-ops | tee polymarket-raw-ops.sha256
             sha256sum binance-usdm-reference-collector \
               | tee binance-usdm-reference-collector.sha256
+            sha256sum binance-usdm-reference-artifact-verifier \
+              | tee binance-usdm-reference-artifact-verifier.sha256
           )
           control_assets=(
             polymarket-raw-ops-shadow-gate.sh
@@ -331,18 +335,22 @@ jobs:
               | awk '{print $1}')
             reference_candidate_sha=$(awk 'NR == 1 {print $1}' \
               binance-usdm-reference-collector.sha256)
+            reference_verifier_sha=$(awk 'NR == 1 {print $1}' \
+              binance-usdm-reference-artifact-verifier.sha256)
             reference_control_manifest_sha=$(sha256sum \
               binance-usdm-reference-control-assets.sha256 | awk '{print $1}')
             reference_control_archive_sha=$(sha256sum \
               binance-usdm-reference-control.tar.gz | awk '{print $1}')
             jq -S -n \
               --arg source_revision '${{ github.sha }}' \
               --arg candidate_sha256 "$reference_candidate_sha" \
+              --arg verifier_sha256 "$reference_verifier_sha" \
               --arg control_manifest_sha256 "$reference_control_manifest_sha" \
               --arg control_archive_sha256 "$reference_control_archive_sha" \
               '{schema:"monday.binance_usdm_reference_release.v1",
                 source_revision:$source_revision,
                 candidate:{file:"binance-usdm-reference-collector",sha256:$candidate_sha256},
+                verifier:{file:"binance-usdm-reference-artifact-verifier",sha256:$verifier_sha256},
                 control_manifest:{file:"binance-usdm-reference-control-assets.sha256",
                   sha256:$control_manifest_sha256},
                 control_archive:{file:"binance-usdm-reference-control.tar.gz",

deployment/aliyun/test-binance-usdm-reference-release-contract.sh

@@ -13,5 +13,10 @@ grep -F -- 'artifact/binance-usdm-reference-collector' "$workflow" >/dev/null
 grep -F -- 'binance-usdm-reference-collector.sha256' "$workflow" >/dev/null
 grep -F -- 'monday.binance_usdm_reference_release.v1' "$workflow" >/dev/null
 grep -F -- 'binance-usdm-reference-release.json.sha256' "$workflow" >/dev/null
+grep -F -- '--bin binance-usdm-reference-artifact-verifier' "$dockerfile" >/dev/null
+grep -F -- '/usr/local/bin/binance-usdm-reference-artifact-verifier' "$dockerfile" >/dev/null
+grep -F -- 'artifact/binance-usdm-reference-artifact-verifier' "$workflow" >/dev/null
+grep -F -- 'binance-usdm-reference-artifact-verifier.sha256' "$workflow" >/dev/null
+grep -F -- 'verifier:{file:"binance-usdm-reference-artifact-verifier"' "$workflow" >/dev/null
 
 printf '%s\n' 'Binance USD-M reference release contract tests passed'

rust_hft/deployment/docker/Dockerfile.binance-lob-archiver

@@ -20,13 +20,16 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     -p hft-collector \
     --bin binance-lob-archiver \
     --bin binance-usdm-reference-collector \
+    --bin binance-usdm-reference-artifact-verifier \
     --bin polymarket-raw-ops \
     && install -D -m 0755 /work/target/release/binance-lob-archiver \
        /out/bin/binance-lob-archiver \
     && install -D -m 0755 /work/target/release/polymarket-raw-ops \
        /out/bin/polymarket-raw-ops \
     && install -D -m 0755 /work/target/release/binance-usdm-reference-collector \
-       /out/bin/binance-usdm-reference-collector
+       /out/bin/binance-usdm-reference-collector \
+    && install -D -m 0755 /work/target/release/binance-usdm-reference-artifact-verifier \
+       /out/bin/binance-usdm-reference-artifact-verifier
 
 FROM debian:bookworm-slim AS aliyun-cli
 
@@ -62,6 +65,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 COPY --from=builder /out/bin/binance-lob-archiver /usr/local/bin/binance-lob-archiver
 COPY --from=builder /out/bin/polymarket-raw-ops /usr/local/bin/polymarket-raw-ops
 COPY --from=builder /out/bin/binance-usdm-reference-collector /usr/local/bin/binance-usdm-reference-collector
+COPY --from=builder /out/bin/binance-usdm-reference-artifact-verifier /usr/local/bin/binance-usdm-reference-artifact-verifier
 COPY --from=aliyun-cli /usr/local/bin/aliyun /usr/local/bin/aliyun
 
 USER hftcollector

rust_hft/tools/collector/Cargo.toml

@@ -88,6 +88,10 @@ path = "src/bin/polymarket-raw-ops.rs"
 name = "binance-usdm-reference-collector"
 path = "src/bin/binance-usdm-reference-collector.rs"
 
+[[bin]]
+name = "binance-usdm-reference-artifact-verifier"
+path = "src/bin/binance-usdm-reference-artifact-verifier.rs"
+
 [lib]
 name = "hft_collector"
 path = "src/lib.rs"

rust_hft/tools/collector/src/bin/binance-usdm-reference-artifact-verifier.rs

@@ -0,0 +1,205 @@
+use anyhow::{bail, Result};
+use clap::Parser;
+use hft_collector::binance_usdm_reference_artifact::{
+    verify_reference_artifact, PublishedReferenceArtifact,
+};
+use serde::Serialize;
+use std::path::{Path, PathBuf};
+
+const OUTPUT_SCHEMA: &str = "monday.binance_usdm_reference_artifact_verification.v1";
+const DATA_NAME: &str = "reference.ndjson";
+
+#[derive(Debug, Parser)]
+#[command(name = "binance-usdm-reference-artifact-verifier")]
+struct Args {
+    #[arg(long)]
+    data_path: PathBuf,
+    #[arg(long)]
+    data_sha256: String,
+    #[arg(long)]
+    manifest_sha256: String,
+}
+
+#[derive(Debug, PartialEq, Eq, Serialize)]
+struct VerificationOutput {
+    schema: &'static str,
+    data_path: PathBuf,
+    data_sha256: String,
+    manifest_sha256: String,
+    metadata_observations: usize,
+    mark_index_funding_observations: usize,
+    open_interest_observations: usize,
+    content_rows_verified: bool,
+}
+
+fn artifact_paths(data_path: &Path) -> Result<(PathBuf, PathBuf)> {
+    if !data_path.is_absolute()
+        || data_path.file_name().and_then(|name| name.to_str()) != Some(DATA_NAME)
+    {
+        bail!("data path must be an absolute canonical reference.ndjson path");
+    }
+    let parent = data_path
+        .parent()
+        .ok_or_else(|| anyhow::anyhow!("data path has no batch directory"))?;
+    Ok((
+        parent.join(format!("{DATA_NAME}.manifest.json")),
+        parent.join(format!("{DATA_NAME}._SUCCESS")),
+    ))
+}
+
+fn verify(args: Args) -> Result<VerificationOutput> {
+    let (manifest_path, success_path) = artifact_paths(&args.data_path)?;
+    let published = PublishedReferenceArtifact {
+        data_path: args.data_path.clone(),
+        manifest_path,
+        success_path,
+        data_sha256: args.data_sha256.clone(),
+        manifest_sha256: args.manifest_sha256.clone(),
+    };
+    let batch = verify_reference_artifact(&published, &args.data_sha256, &args.manifest_sha256)?;
+    Ok(VerificationOutput {
+        schema: OUTPUT_SCHEMA,
+        data_path: args.data_path,
+        data_sha256: args.data_sha256,
+        manifest_sha256: args.manifest_sha256,
+        metadata_observations: batch.contracts().len(),
+        mark_index_funding_observations: batch.mark_index_funding().len(),
+        open_interest_observations: batch.open_interest().len(),
+        content_rows_verified: true,
+    })
+}
+
+fn main() -> Result<()> {
+    let output = verify(Args::parse())?;
+    println!("{}", serde_json::to_string(&output)?);
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use data::binance_usdm_reference::{
+        ActivePerpetualContract, CompleteReferenceBatch, MarkIndexFundingObservation,
+        OpenInterestObservation, EXCHANGE_INFO_ENDPOINT, OPEN_INTEREST_ENDPOINT,
+        PREMIUM_INDEX_ENDPOINT, REFERENCE_SCHEMA, SERVER_TIME_ENDPOINT,
+    };
+    use hft_collector::binance_usdm_reference_artifact::{
+        publish_reference_batch, ReferenceArtifactConfig,
+    };
+    use hft_collector::binance_usdm_reference_collector::OFFICIAL_USDM_SOURCE_ORIGIN;
+    use rust_decimal::Decimal;
+    use std::fs;
+    use std::os::unix::fs::symlink;
+    use tempfile::tempdir;
+
+    const SOURCE_MS: u64 = 1_700_000_000_000;
+    const RECEIVED_NS: u64 = 1_700_000_000_500_000_000;
+
+    fn sample_batch() -> CompleteReferenceBatch {
+        CompleteReferenceBatch::new(
+            vec![ActivePerpetualContract {
+                schema: REFERENCE_SCHEMA.to_owned(),
+                symbol: "BTCUSDT".to_owned(),
+                pair: "BTCUSDT".to_owned(),
+                base_asset: "BTC".to_owned(),
+                quote_asset: "USDT".to_owned(),
+                margin_asset: "USDT".to_owned(),
+                contract_type: "PERPETUAL".to_owned(),
+                status: "TRADING".to_owned(),
+                onboard_date_ms: 1,
+                delivery_date_ms: 4_133_404_800_000,
+                source_time_ms: SOURCE_MS,
+                source_clock_received_at_ns: RECEIVED_NS - 100,
+                received_at_ns: RECEIVED_NS - 50,
+                source_endpoint: EXCHANGE_INFO_ENDPOINT.to_owned(),
+                source_clock_endpoint: SERVER_TIME_ENDPOINT.to_owned(),
+            }],
+            vec![MarkIndexFundingObservation {
+                schema: REFERENCE_SCHEMA.to_owned(),
+                symbol: "BTCUSDT".to_owned(),
+                mark_price: Decimal::new(101, 0),
+                index_price: Decimal::new(100, 0),
+                basis: Decimal::ONE,
+                basis_rate: Decimal::new(1, 2),
+                last_funding_rate: Decimal::new(1, 4),
+                interest_rate: Decimal::new(1, 4),
+                next_funding_time_ms: SOURCE_MS + 28_800_000,
+                source_time_ms: SOURCE_MS,
+                received_at_ns: RECEIVED_NS,
+                source_endpoint: PREMIUM_INDEX_ENDPOINT.to_owned(),
+            }],
+            vec![OpenInterestObservation {
+                schema: REFERENCE_SCHEMA.to_owned(),
+                symbol: "BTCUSDT".to_owned(),
+                open_interest: Decimal::new(12345, 3),
+                source_time_ms: SOURCE_MS,
+                received_at_ns: RECEIVED_NS + 50,
+                source_endpoint: OPEN_INTEREST_ENDPOINT.to_owned(),
+            }],
+        )
+        .unwrap()
+    }
+
+    fn fixture() -> (tempfile::TempDir, PublishedReferenceArtifact) {
+        let temp = tempdir().unwrap();
+        let output_root = fs::canonicalize(temp.path()).unwrap();
+        let published = publish_reference_batch(
+            &ReferenceArtifactConfig {
+                output_root,
+                observed_at_ns: RECEIVED_NS + 100,
+                max_staleness_ms: 1_000,
+            },
+            OFFICIAL_USDM_SOURCE_ORIGIN,
+            &sample_batch(),
+        )
+        .unwrap();
+        (temp, published)
+    }
+
+    fn args(published: &PublishedReferenceArtifact) -> Args {
+        Args {
+            data_path: published.data_path.clone(),
+            data_sha256: published.data_sha256.clone(),
+            manifest_sha256: published.manifest_sha256.clone(),
+        }
+    }
+
+    #[test]
+    fn verifies_canonical_triplet_and_reports_content_counts() {
+        let (_temp, published) = fixture();
+        let output = verify(args(&published)).unwrap();
+        assert_eq!(output.schema, OUTPUT_SCHEMA);
+        assert_eq!(output.metadata_observations, 1);
+        assert_eq!(output.mark_index_funding_observations, 1);
+        assert_eq!(output.open_interest_observations, 1);
+        assert!(output.content_rows_verified);
+    }
+
+    #[test]
+    fn rejects_tampered_data_even_when_the_caller_supplies_the_old_anchor() {
+        let (_temp, published) = fixture();
+        fs::write(&published.data_path, b"{\"tampered\":true}\n").unwrap();
+        assert!(verify(args(&published)).is_err());
+    }
+
+    #[test]
+    fn rejects_symlinked_manifest() {
+        let (_temp, published) = fixture();
+        let real = published.manifest_path.with_extension("json.real");
+        fs::rename(&published.manifest_path, &real).unwrap();
+        symlink(&real, &published.manifest_path).unwrap();
+        assert!(verify(args(&published)).is_err());
+    }
+
+    #[test]
+    fn rejects_noncanonical_or_wrongly_anchored_inputs() {
+        let (_temp, published) = fixture();
+        let mut relative = args(&published);
+        relative.data_path = PathBuf::from("reference.ndjson");
+        assert!(verify(relative).is_err());
+
+        let mut wrong = args(&published);
+        wrong.data_sha256 = "0".repeat(64);
+        assert!(verify(wrong).is_err());
+    }
+}