Update mobile verifiers to use new info_at_timestamp API

mobile_config/src/gateway/client.rs

@@ -1,5 +1,6 @@
 use crate::client::{call_with_retry, ClientError, Settings};
 use crate::gateway::service::info::{GatewayInfo, GatewayInfoStream};
+use chrono::{DateTime, Utc};
 use file_store_oracles::traits::MsgVerify;
 use futures::stream::{self, StreamExt};
 use helium_crypto::{Keypair, PublicKey, PublicKeyBinary, Sign};
@@ -49,6 +50,7 @@ pub trait GatewayInfoResolver: Clone + Send + Sync + 'static {
     async fn resolve_gateway_info(
         &self,
         address: &PublicKeyBinary,
+        gateway_query_timestamp: &DateTime<Utc>,
     ) -> Result<Option<GatewayInfo>, ClientError>;
 
     async fn stream_gateways_info(
@@ -62,27 +64,30 @@ impl GatewayInfoResolver for GatewayClient {
     async fn resolve_gateway_info(
         &self,
         address: &PublicKeyBinary,
+        gateway_query_timestamp: &DateTime<Utc>,
     ) -> Result<Option<GatewayInfo>, ClientError> {
         if let Some(cached_response) = self.cache.get(address).await {
             return Ok(cached_response.value().clone());
         }
 
-        let mut request = mobile_config::GatewayInfoReqV1 {
+        let mut request = mobile_config::GatewayInfoAtTimestampReqV1 {
             address: address.clone().into(),
             signer: self.signing_key.public_key().into(),
             signature: vec![],
+            query_time: gateway_query_timestamp.clone().timestamp() as u64,
         };
         request.signature = self.signing_key.sign(&request.encode_to_vec())?;
         tracing::debug!(pubkey = address.to_string(), "fetching gateway info");
-        let response = match call_with_retry!(self.client.clone().info_v2(request.clone())) {
-            Ok(info_res) => {
-                let response = info_res.into_inner();
-                response.verify(&self.config_pubkey)?;
-                response.info.map(GatewayInfo::try_from).transpose()?
-            }
-            Err(status) if status.code() == tonic::Code::NotFound => None,
-            Err(status) => Err(status)?,
-        };
+        let response =
+            match call_with_retry!(self.client.clone().info_at_timestamp(request.clone())) {
+                Ok(info_res) => {
+                    let response = info_res.into_inner();
+                    response.verify(&self.config_pubkey)?;
+                    response.info.map(GatewayInfo::try_from).transpose()?
+                }
+                Err(status) if status.code() == tonic::Code::NotFound => None,
+                Err(status) => Err(status)?,
+            };
 
         self.cache
             .insert(address.clone(), response.clone(), self.cache_ttl)

mobile_packet_verifier/src/accumulate.rs

@@ -79,7 +79,10 @@ async fn verify_report(
         return Ok(ReportStatus::Banned);
     }
 
-    if !mobile_config.is_gateway_known(gw_pub_key).await {
+    if !mobile_config
+        .is_gateway_known(gw_pub_key, &report.received_timestamp)
+        .await
+    {
         return Ok(ReportStatus::InvalidGatewayKey);
     }
 

mobile_packet_verifier/src/banning/mod.rs

@@ -21,7 +21,7 @@ pub const BAN_CLEANUP_DAYS: i64 = 7;
 #[derive(Debug, Deserialize, Serialize)]
 pub struct BanSettings {
     /// Where do we look in s3 for ban files
-    pub input_bucket: String,
+    pub input_bucket: file_store::BucketSettings,
     /// How often to purge expired bans
     #[serde(with = "humantime_serde", default = "default_purge_interval")]
     pub purge_interval: Duration,
@@ -40,12 +40,11 @@ fn default_ingest_start_after() -> DateTime<Utc> {
 
 pub async fn create_managed_task(
     pool: PgPool,
-    client: file_store::Client,
     settings: &BanSettings,
 ) -> anyhow::Result<impl ManagedTask> {
     let (ban_report_rx, ban_report_server) = file_source::continuous_source()
         .state(pool.clone())
-        .file_store(client, settings.input_bucket.clone())
+        .bucket_client(settings.input_bucket.connect().await)
         .lookback_start_after(settings.start_after)
         .prefix(FileType::VerifiedMobileBanReport.to_string())
         .create()

mobile_packet_verifier/src/daemon.rs

@@ -156,9 +156,8 @@ impl Cmd {
             None
         };
 
-        let file_store_client = settings.file_store.connect().await;
         let (file_upload, file_upload_server) =
-            file_upload::FileUpload::new(file_store_client.clone(), settings.output_bucket.clone())
+            file_upload::FileUpload::from_bucket_client(settings.output_bucket.connect().await)
                 .await;
 
         let (valid_sessions, valid_sessions_server) = ValidDataTransferSession::file_sink(
@@ -189,7 +188,7 @@ impl Cmd {
 
         let (reports, reports_server) = file_source::continuous_source()
             .state(pool.clone())
-            .file_store(file_store_client.clone(), settings.ingest_bucket.clone())
+            .bucket_client(settings.ingest_bucket.connect().await)
             .prefix(FileType::DataTransferSessionIngestReport.to_string())
             .lookback_start_after(settings.start_after)
             .create()
@@ -207,8 +206,7 @@ impl Cmd {
         );
 
         let event_id_purger = EventIdPurger::from_settings(pool.clone(), settings);
-        let banning =
-            banning::create_managed_task(pool, file_store_client, &settings.banning).await?;
+        let banning = banning::create_managed_task(pool, &settings.banning).await?;
 
         TaskManager::builder()
             .add_task(file_upload_server)