Skip to content

Commit

Permalink
Revert PR 774 so it can continue to be tested (#777)
Browse files Browse the repository at this point in the history 
* Revert "Fix SQL (#776)"

This reverts commit 40e6541.

* Revert "Lookup previous WiFi locations in case of None location validation timestamp  (#774)"

This reverts commit fab10f6.
  • Loading branch information
@bbalser
bbalser authored Mar 28, 2024
1 parent 40e6541 commit 317e731
Show file tree
Hide file tree
Showing 6 changed files with 14 additions and 182 deletions.
3 changes: 0 additions & 3 deletions mobile_verifier/migrations/30_save_lat_and_lon.sql
View file

This file was deleted.

8 changes: 1 addition & 7 deletions mobile_verifier/src/heartbeats/cbrs.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ use super::{process_validated_heartbeats, Heartbeat, ValidatedHeartbeat};
use crate::{
coverage::{CoverageClaimTimeCache, CoverageObjectCache},
geofence::GeofenceValidator,
heartbeats::LocationCache,
GatewayResolver,
};

Expand Down Expand Up @@ -75,8 +74,6 @@ where

let coverage_claim_time_cache = CoverageClaimTimeCache::new();
let coverage_object_cache = CoverageObjectCache::new(&self.pool);
// Unused:
let location_cache = LocationCache::new(&self.pool);

loop {
#[rustfmt::skip]
Expand All @@ -93,7 +90,6 @@ where
&heartbeat_cache,
&coverage_claim_time_cache,
&coverage_object_cache,
&location_cache,
).await?;
metrics::histogram!("cbrs_heartbeat_processing_time", start.elapsed());
}
Expand All @@ -109,7 +105,6 @@ where
heartbeat_cache: &Arc<Cache<(String, DateTime<Utc>), ()>>,
coverage_claim_time_cache: &CoverageClaimTimeCache,
coverage_object_cache: &CoverageObjectCache,
location_cache: &LocationCache,
) -> anyhow::Result<()> {
tracing::info!("Processing CBRS heartbeat file {}", file.file_info.key);
let mut transaction = self.pool.begin().await?;
Expand All @@ -127,10 +122,9 @@ where
});
process_validated_heartbeats(
ValidatedHeartbeat::validate_heartbeats(
heartbeats,
&self.gateway_info_resolver,
heartbeats,
coverage_object_cache,
location_cache,
self.max_distance_to_asserted,
self.max_distance_to_coverage,
&epoch,
Expand Down
159 changes: 8 additions & 151 deletions mobile_verifier/src/heartbeats/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@ use helium_proto::services::poc_mobile as proto;
use retainer::Cache;
use rust_decimal::{prelude::ToPrimitive, Decimal};
use rust_decimal_macros::dec;
use sqlx::{postgres::PgTypeInfo, Decode, Encode, PgPool, Postgres, Transaction, Type};
use std::{ops::Range, pin::pin, sync::Arc, time};
use sqlx::{postgres::PgTypeInfo, Decode, Encode, Postgres, Transaction, Type};
use std::{ops::Range, pin::pin, time};
use uuid::Uuid;

/// Minimum number of heartbeats required to give a reward to the hotspot.
Expand Down Expand Up @@ -379,12 +379,10 @@ impl ValidatedHeartbeat {
}

/// Validate a heartbeat in the given epoch.
#[allow(clippy::too_many_arguments)]
pub async fn validate(
mut heartbeat: Heartbeat,
heartbeat: Heartbeat,
gateway_info_resolver: &impl GatewayResolver,
coverage_object_cache: &CoverageObjectCache,
last_location_cache: &LocationCache,
max_distance_to_asserted: u32,
max_distance_to_coverage: u32,
epoch: &Range<DateTime<Utc>>,
Expand Down Expand Up @@ -472,7 +470,7 @@ impl ValidatedHeartbeat {
));
}

let Ok(mut hb_latlng) = heartbeat.centered_latlng() else {
let Ok(hb_latlng) = heartbeat.centered_latlng() else {
return Ok(Self::new(
heartbeat,
cell_type,
Expand Down Expand Up @@ -518,39 +516,8 @@ impl ValidatedHeartbeat {
}
GatewayResolution::AssertedLocation(location) if heartbeat.hb_type == HbType::Wifi => {
let asserted_latlng: LatLng = CellIndex::try_from(location)?.into();
let is_valid = match heartbeat.location_validation_timestamp {
None => {
if let Some(last_location) = last_location_cache
.fetch_last_location(&heartbeat.hotspot_key)
.await?
{
heartbeat.lat = last_location.lat;
heartbeat.lon = last_location.lon;
heartbeat.location_validation_timestamp =
Some(last_location.location_validation_timestamp);
// Can't panic, previous lat and lon must be valid.
hb_latlng = heartbeat.centered_latlng().unwrap();
true
} else {
false
}
}
Some(location_validation_timestamp) => {
last_location_cache
.set_last_location(
&heartbeat.hotspot_key,
LastLocation::new(
location_validation_timestamp,
heartbeat.lat,
heartbeat.lon,
),
)
.await?;
true
}
};
let distance_to_asserted = asserted_latlng.distance_m(hb_latlng).round() as i64;
let location_trust_score_multiplier = if is_valid
let location_trust_score_multiplier = if heartbeat.location_validation_timestamp.is_some()
// The heartbeat location to asserted location must be less than the max_distance_to_asserted value:
&& distance_to_asserted <= max_distance_to_asserted as i64
// The heartbeat location to every associated coverage hex must be less than max_distance_to_coverage:
Expand Down Expand Up @@ -580,12 +547,10 @@ impl ValidatedHeartbeat {
}
}

#[allow(clippy::too_many_arguments)]
pub fn validate_heartbeats<'a>(
heartbeats: impl Stream<Item = Heartbeat> + 'a,
gateway_info_resolver: &'a impl GatewayResolver,
heartbeats: impl Stream<Item = Heartbeat> + 'a,
coverage_object_cache: &'a CoverageObjectCache,
last_location_cache: &'a LocationCache,
max_distance_to_asserted: u32,
max_distance_to_coverage: u32,
epoch: &'a Range<DateTime<Utc>>,
Expand All @@ -596,7 +561,6 @@ impl ValidatedHeartbeat {
heartbeat,
gateway_info_resolver,
coverage_object_cache,
last_location_cache,
max_distance_to_asserted,
max_distance_to_coverage,
epoch,
Expand Down Expand Up @@ -690,8 +654,8 @@ impl ValidatedHeartbeat {
let truncated_timestamp = self.truncated_timestamp()?;
sqlx::query(
r#"
INSERT INTO wifi_heartbeats (hotspot_key, cell_type, latest_timestamp, truncated_timestamp, coverage_object, location_trust_score_multiplier, distance_to_asserted, lat, lon)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
INSERT INTO wifi_heartbeats (hotspot_key, cell_type, latest_timestamp, truncated_timestamp, coverage_object, location_trust_score_multiplier, distance_to_asserted)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (hotspot_key, truncated_timestamp) DO UPDATE SET
latest_timestamp = EXCLUDED.latest_timestamp,
coverage_object = EXCLUDED.coverage_object
Expand All @@ -704,8 +668,6 @@ impl ValidatedHeartbeat {
.bind(self.heartbeat.coverage_object)
.bind(self.location_trust_score_multiplier)
.bind(self.distance_to_asserted)
.bind(self.heartbeat.lat)
.bind(self.heartbeat.lon)
.execute(&mut *exec)
.await?;
Ok(())
Expand Down Expand Up @@ -779,111 +741,6 @@ pub async fn clear_heartbeats(
Ok(())
}

/// A cache for previous valid (or invalid) WiFi heartbeat locations
#[derive(Clone)]
pub struct LocationCache {
pool: PgPool,
locations: Arc<Cache<PublicKeyBinary, Option<LastLocation>>>,
}

impl LocationCache {
pub fn new(pool: &PgPool) -> Self {
let locations = Arc::new(Cache::new());
let locations_clone = locations.clone();
tokio::spawn(async move {
locations_clone
.monitor(4, 0.25, std::time::Duration::from_secs(60 * 60 * 24))
.await
});
Self {
pool: pool.clone(),
locations,
}
}

async fn fetch_from_db_and_set(
&self,
hotspot: &PublicKeyBinary,
) -> anyhow::Result<Option<LastLocation>> {
let last_location: Option<LastLocation> = sqlx::query_as(
r#"
SELECT location_validation_timestamp, lat, lon
FROM wifi_heartbeats
WHERE location_validation_timestamp IS NOT NULL
AND location_validation_timestamp >= $1
ORDER BY location_validation_timestamp DESC
LIMIT 1
"#,
)
.bind(Utc::now() - Duration::hours(12))
.fetch_optional(&self.pool)
.await?;
self.locations
.insert(
hotspot.clone(),
last_location,
last_location
.map(|x| x.duration_to_expiration())
.unwrap_or_else(|| Duration::days(365))
.to_std()?,
)
.await;
Ok(last_location)
}

pub async fn fetch_last_location(
&self,
hotspot: &PublicKeyBinary,
) -> anyhow::Result<Option<LastLocation>> {
Ok(
if let Some(last_location) = self.locations.get(hotspot).await {
*last_location
} else {
self.fetch_from_db_and_set(hotspot).await?
},
)
}

pub async fn set_last_location(
&self,
hotspot: &PublicKeyBinary,
last_location: LastLocation,
) -> anyhow::Result<()> {
let duration_to_expiration = last_location.duration_to_expiration();
self.locations
.insert(
hotspot.clone(),
Some(last_location),
duration_to_expiration.to_std()?,
)
.await;
Ok(())
}
}

#[derive(sqlx::FromRow, Copy, Clone)]
pub struct LastLocation {
location_validation_timestamp: DateTime<Utc>,
lat: f64,
lon: f64,
}

impl LastLocation {
fn new(location_validation_timestamp: DateTime<Utc>, lat: f64, lon: f64) -> Self {
Self {
location_validation_timestamp,
lat,
lon,
}
}

/// Calculates the duration from now in which last_valid_timestamp is 12 hours old
fn duration_to_expiration(&self) -> Duration {
((self.location_validation_timestamp + Duration::hours(12)) - Utc::now())
.max(Duration::zero())
}
}

pub struct SeniorityUpdate<'a> {
heartbeat: &'a ValidatedHeartbeat,
action: SeniorityUpdateAction,
Expand Down
7 changes: 1 addition & 6 deletions mobile_verifier/src/heartbeats/wifi.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ use super::{process_validated_heartbeats, Heartbeat, ValidatedHeartbeat};
use crate::{
coverage::{CoverageClaimTimeCache, CoverageObjectCache},
geofence::GeofenceValidator,
heartbeats::LocationCache,
GatewayResolver,
};
use chrono::{DateTime, Duration, Utc};
Expand Down Expand Up @@ -74,7 +73,6 @@ where

let coverage_claim_time_cache = CoverageClaimTimeCache::new();
let coverage_object_cache = CoverageObjectCache::new(&self.pool);
let location_cache = LocationCache::new(&self.pool);

loop {
#[rustfmt::skip]
Expand All @@ -91,7 +89,6 @@ where
&heartbeat_cache,
&coverage_claim_time_cache,
&coverage_object_cache,
&location_cache
).await?;
metrics::histogram!("wifi_heartbeat_processing_time", start.elapsed());
}
Expand All @@ -107,7 +104,6 @@ where
heartbeat_cache: &Cache<(String, DateTime<Utc>), ()>,
coverage_claim_time_cache: &CoverageClaimTimeCache,
coverage_object_cache: &CoverageObjectCache,
location_cache: &LocationCache,
) -> anyhow::Result<()> {
tracing::info!("Processing WIFI heartbeat file {}", file.file_info.key);
let mut transaction = self.pool.begin().await?;
Expand All @@ -119,10 +115,9 @@ where
.map(Heartbeat::from);
process_validated_heartbeats(
ValidatedHeartbeat::validate_heartbeats(
heartbeats,
&self.gateway_info_resolver,
heartbeats,
coverage_object_cache,
location_cache,
self.max_distance_to_asserted,
self.max_distance_to_coverage,
&epoch,
Expand Down
6 changes: 2 additions & 4 deletions mobile_verifier/tests/boosting_oracles.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use mobile_verifier::{
CoverageObjectCache, Seniority, UnassignedHex,
},
geofence::GeofenceValidator,
heartbeats::{Heartbeat, HeartbeatReward, LocationCache, SeniorityUpdate, ValidatedHeartbeat},
heartbeats::{Heartbeat, HeartbeatReward, SeniorityUpdate, ValidatedHeartbeat},
radio_threshold::VerifiedRadioThresholds,
reward_shares::CoveragePoints,
speedtests::Speedtest,
Expand Down Expand Up @@ -371,14 +371,12 @@ async fn test_footfall_and_urbanization(pool: PgPool) -> anyhow::Result<()> {

let coverage_objects = CoverageObjectCache::new(&pool);
let coverage_claim_time_cache = CoverageClaimTimeCache::new();
let location_cache = LocationCache::new(&pool);

let epoch = start..end;
let mut heartbeats = pin!(ValidatedHeartbeat::validate_heartbeats(
stream::iter(heartbeats.map(Heartbeat::from)),
&AllOwnersValid,
stream::iter(heartbeats.map(Heartbeat::from)),
&coverage_objects,
&location_cache,
2000,
2000,
&epoch,
Expand Down
Loading

0 comments on commit 317e731

Please sign in to comment.