Merge pull request #784 from helium/andymck/one-beacon-attempt-per-wi…

…ndow

track last beacon ts for schedule and recip separately. ensure only one beacon attempt per beaconing window

iot_verifier/migrations/15_last_beacon_reciprocity.sql

@@ -0,0 +1,8 @@
+create table last_beacon_recip (
+    id bytea primary key not null,
+    timestamp timestamptz not null
+);
+-- seed beacon_recip with timestamps from last_beacon
+insert into last_beacon_recip (id, timestamp)
+select id, timestamp from last_beacon
+where timestamp > now() - interval '7 day';

iot_verifier/src/last_beacon.rs

@@ -1,7 +1,7 @@
 use chrono::{DateTime, Utc};
 use helium_crypto::PublicKeyBinary;
 use serde::{Deserialize, Serialize};
-use sqlx::{postgres::PgRow, FromRow, Row};
+use sqlx::{postgres::PgRow, FromRow, Postgres, Row, Transaction};
 
 #[derive(Deserialize, Serialize, Debug)]
 pub struct LastBeacon {
@@ -86,14 +86,11 @@ impl LastBeacon {
         Ok(height)
     }
 
-    pub async fn update_last_timestamp<'c, E>(
-        executor: E,
+    pub async fn update_last_timestamp(
+        txn: &mut Transaction<'_, Postgres>,
         id: &PublicKeyBinary,
         timestamp: DateTime<Utc>,
-    ) -> anyhow::Result<()>
-    where
-        E: sqlx::Executor<'c, Database = sqlx::Postgres>,
-    {
+    ) -> anyhow::Result<()> {
         let _ = sqlx::query(
             r#"
             insert into last_beacon (id, timestamp)
@@ -104,7 +101,7 @@ impl LastBeacon {
         )
         .bind(id.as_ref())
         .bind(timestamp)
-        .execute(executor)
+        .execute(txn)
         .await?;
         Ok(())
     }

iot_verifier/src/last_beacon_reciprocity.rs

@@ -0,0 +1,53 @@
+use chrono::{DateTime, Utc};
+use helium_crypto::PublicKeyBinary;
+use serde::{Deserialize, Serialize};
+use sqlx::{postgres::PgRow, FromRow, Postgres, Row, Transaction};
+
+#[derive(Deserialize, Serialize, Debug)]
+pub struct LastBeaconReciprocity {
+    pub id: PublicKeyBinary,
+    pub timestamp: DateTime<Utc>,
+}
+
+impl FromRow<'_, PgRow> for LastBeaconReciprocity {
+    fn from_row(row: &PgRow) -> sqlx::Result<Self> {
+        Ok(Self {
+            id: row.get::<Vec<u8>, &str>("id").into(),
+            timestamp: row.get::<DateTime<Utc>, &str>("timestamp"),
+        })
+    }
+}
+
+impl LastBeaconReciprocity {
+    pub async fn get<'c, E>(executor: E, id: &PublicKeyBinary) -> anyhow::Result<Option<Self>>
+    where
+        E: sqlx::Executor<'c, Database = Postgres>,
+    {
+        Ok(sqlx::query_as::<_, LastBeaconReciprocity>(
+            r#" select * from last_beacon_recip where id = $1;"#,
+        )
+        .bind(id.as_ref())
+        .fetch_optional(executor)
+        .await?)
+    }
+
+    pub async fn update_last_timestamp(
+        txn: &mut Transaction<'_, Postgres>,
+        id: &PublicKeyBinary,
+        timestamp: DateTime<Utc>,
+    ) -> anyhow::Result<()> {
+        let _ = sqlx::query(
+            r#"
+            insert into last_beacon_recip (id, timestamp)
+            values ($1, $2)
+            on conflict (id) do update set
+                timestamp = EXCLUDED.timestamp
+            "#,
+        )
+        .bind(id.as_ref())
+        .bind(timestamp)
+        .execute(txn)
+        .await?;
+        Ok(())
+    }
+}

iot_verifier/src/lib.rs

@@ -4,6 +4,7 @@ pub mod gateway_cache;
 pub mod gateway_updater;
 pub mod hex_density;
 pub mod last_beacon;
+pub mod last_beacon_reciprocity;
 pub mod last_witness;
 pub mod loader;
 pub mod meta;

iot_verifier/src/poc.rs

@@ -3,6 +3,7 @@ use crate::{
     gateway_cache::{GatewayCache, GatewayCacheError},
     hex_density::HexDensityMap,
     last_beacon::LastBeacon,
+    last_beacon_reciprocity::LastBeaconReciprocity,
     last_witness::LastWitness,
     region_cache::RegionCache,
     witness_updater::WitnessUpdater,
@@ -148,7 +149,7 @@ impl Poc {
         };
         // we have beaconer info, proceed to verifications
         let last_beacon = LastBeacon::get(&self.pool, &beaconer_pub_key).await?;
-        match do_beacon_verifications(
+        let result = match do_beacon_verifications(
             deny_list,
             self.entropy_start,
             self.entropy_end,
@@ -164,24 +165,35 @@ impl Poc {
                     .get(beaconer_metadata.location)
                     .await
                     .unwrap_or(*DEFAULT_TX_SCALE);
-                // update 'last beacon' timestamp if the beacon has passed regular validations
-                // but only if there has been at least one witness report
-                if !self.witness_reports.is_empty() {
-                    LastBeacon::update_last_timestamp(
-                        &self.pool,
-                        &beaconer_pub_key,
-                        self.beacon_report.received_timestamp,
-                    )
-                    .await?
-                }
-                Ok(VerifyBeaconResult::valid(beaconer_info, tx_scale))
+                VerifyBeaconResult::valid(beaconer_info, tx_scale)
             }
-            Err(invalid_response) => Ok(VerifyBeaconResult::invalid(
+            Err(invalid_response) => VerifyBeaconResult::invalid(
                 invalid_response.reason,
                 invalid_response.details,
                 beaconer_info,
-            )),
+            ),
+        };
+        let mut txn = self.pool.begin().await?;
+        // update 'last beacon' timestamp irrespective of whether the beacon is valid or not
+        LastBeacon::update_last_timestamp(
+            &mut txn,
+            &beaconer_pub_key,
+            self.beacon_report.received_timestamp,
+        )
+        .await?;
+        // update 'last beacon reciprocity' timestamp if the beacon has passed regular validations
+        // and has at least one witness report
+        if result.result == VerificationStatus::Valid && !self.witness_reports.is_empty() {
+            LastBeaconReciprocity::update_last_timestamp(
+                &mut txn,
+                &beaconer_pub_key,
+                self.beacon_report.received_timestamp,
+            )
+            .await?
         }
+        txn.commit().await?;
+
+        Ok(result)
     }
 
     pub async fn verify_witnesses(

iot_verifier/src/runner.rs

@@ -1,7 +1,7 @@
 use crate::{
     gateway_cache::GatewayCache,
     hex_density::HexDensityMap,
-    last_beacon::LastBeacon,
+    last_beacon_reciprocity::LastBeaconReciprocity,
     poc::{Poc, VerifyBeaconResult},
     poc_report::Report,
     region_cache::RegionCache,
@@ -567,8 +567,9 @@ where
         &self,
         report: &IotVerifiedWitnessReport,
     ) -> anyhow::Result<bool> {
-        let last_beacon = LastBeacon::get(&self.pool, &report.report.pub_key).await?;
-        Ok(last_beacon.map_or(false, |lw| {
+        let last_beacon_recip =
+            LastBeaconReciprocity::get(&self.pool, &report.report.pub_key).await?;
+        Ok(last_beacon_recip.map_or(false, |lw| {
             report.received_timestamp - lw.timestamp < *RECIPROCITY_WINDOW
         }))
     }

iot_verifier/tests/common/mod.rs

@@ -21,7 +21,7 @@ use iot_config::{
 };
 use iot_verifier::{
     entropy::Entropy,
-    last_beacon::LastBeacon,
+    last_beacon_reciprocity::LastBeaconReciprocity,
     last_witness::LastWitness,
     poc_report::{InsertBindings, IotStatus, Report, ReportType},
 };
@@ -311,7 +311,7 @@ pub async fn inject_last_beacon(
     gateway: PublicKeyBinary,
     ts: DateTime<Utc>,
 ) -> anyhow::Result<()> {
-    LastBeacon::update_last_timestamp(&mut *txn, &gateway, ts).await
+    LastBeaconReciprocity::update_last_timestamp(&mut *txn, &gateway, ts).await
 }
 
 #[allow(dead_code)]