Correct FileSink auto_commit and roll_time usage in Oracles

The PR that updated FileSink to use the Ext trait had an incorrect understanding of the relationship between `roll_time` and `auto_commit`. That understanding has since been corrected and codified in the `FileSinkCommitStrategy` enum. For this commit, https://github.com/helium/oracles/pull/849/files was combed through to get all correct settings for FileSinks before they were transitioned to use FileSinkWriteExt::file_sink().
helium · Aug 29, 2024 · f96a2cd · f96a2cd
1 parent 0faed88
commit f96a2cd
Show file tree

Hide file tree

Showing 17 changed files with 72 additions and 59 deletions.
diff --git a/boost_manager/src/main.rs b/boost_manager/src/main.rs
@@ -5,8 +5,11 @@ use boost_manager::{
 };
 use clap::Parser;
 use file_store::{
-    file_info_poller::LookbackBehavior, file_source, file_upload, reward_manifest::RewardManifest,
-    traits::FileSinkWriteExt, FileStore, FileType,
+    file_info_poller::LookbackBehavior,
+    file_source, file_upload,
+    reward_manifest::RewardManifest,
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
+    FileStore, FileType,
 };
 use helium_proto::BoostedHexUpdateV1;
 use mobile_config::client::hex_boosting_client::HexBoostingClient;
@@ -103,7 +106,7 @@ impl Server {
         let (updated_hexes_sink, updated_hexes_sink_server) = BoostedHexUpdateV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(Duration::from_secs(5 * 60)),
+            FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/ingest/src/server_iot.rs b/ingest/src/server_iot.rs
@@ -4,7 +4,7 @@ use chrono::Utc;
 use file_store::{
     file_sink::FileSinkClient,
     file_upload,
-    traits::{FileSinkWriteExt, MsgVerify},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt, MsgVerify},
 };
 use futures::{
     future::{LocalBoxFuture, TryFutureExt},
@@ -365,7 +365,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
     let (beacon_report_sink, beacon_report_sink_server) = LoraBeaconIngestReportV1::file_sink(
         store_base_path,
         file_upload.clone(),
-        Some(Duration::from_secs(5 * 60)),
+        FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
         env!("CARGO_PKG_NAME"),
     )
     .await?;
@@ -374,7 +374,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
     let (witness_report_sink, witness_report_sink_server) = LoraWitnessIngestReportV1::file_sink(
         store_base_path,
         file_upload.clone(),
-        Some(Duration::from_secs(5 * 60)),
+        FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
         env!("CARGO_PKG_NAME"),
     )
     .await?;

diff --git a/ingest/src/server_mobile.rs b/ingest/src/server_mobile.rs
@@ -4,7 +4,7 @@ use chrono::Utc;
 use file_store::{
     file_sink::FileSinkClient,
     file_upload,
-    traits::{FileSinkWriteExt, MsgVerify},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt, MsgVerify},
 };
 use futures::future::LocalBoxFuture;
 use futures_util::TryFutureExt;
@@ -450,7 +450,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         CellHeartbeatIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -459,7 +459,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         WifiHeartbeatIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -468,7 +468,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
     let (speedtest_report_sink, speedtest_report_sink_server) = SpeedtestIngestReportV1::file_sink(
         store_base_path,
         file_upload.clone(),
-        Some(settings.roll_time),
+        FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
         env!("CARGO_PKG_NAME"),
     )
     .await?;
@@ -477,7 +477,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         DataTransferSessionIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -486,7 +486,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         SubscriberLocationIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -495,7 +495,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         RadioThresholdIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -504,7 +504,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         InvalidatedRadioThresholdIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -513,7 +513,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         CoverageObjectIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -522,7 +522,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         ServiceProviderBoostedRewardsBannedRadioIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -531,7 +531,7 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         SubscriberVerifiedMappingEventIngestReportV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(settings.roll_time),
+            FileSinkCommitStrategy::AutomaticRollTime(settings.roll_time),
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/iot_packet_verifier/src/daemon.rs b/iot_packet_verifier/src/daemon.rs
@@ -11,7 +11,7 @@ use file_store::{
     file_sink::FileSinkClient,
     file_source, file_upload,
     iot_packet::PacketRouterPacketReport,
-    traits::{FileSinkWriteExt, DEFAULT_ROLL_TIME},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
     FileStore, FileType,
 };
 use futures_util::TryFutureExt;
@@ -141,15 +141,15 @@ impl Cmd {
         let (valid_packets, valid_packets_server) = ValidPacket::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(DEFAULT_ROLL_TIME),
+            FileSinkCommitStrategy::Manual,
             env!("CARGO_PKG_NAME"),
         )
         .await?;
 
         let (invalid_packets, invalid_packets_server) = InvalidPacket::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(DEFAULT_ROLL_TIME),
+            FileSinkCommitStrategy::Manual,
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/iot_verifier/src/main.rs b/iot_verifier/src/main.rs
@@ -6,7 +6,7 @@ use file_store::{
     file_info_poller::LookbackBehavior,
     file_source, file_upload,
     iot_packet::IotValidPacket,
-    traits::{FileSinkWriteExt, DEFAULT_ROLL_TIME},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
     FileStore, FileType,
 };
 use helium_proto::{
@@ -124,7 +124,7 @@ impl Server {
         let (rewards_sink, gateway_rewards_sink_server) = IotRewardShare::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(DEFAULT_ROLL_TIME),
+            FileSinkCommitStrategy::Manual,
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -133,7 +133,7 @@ impl Server {
         let (reward_manifests_sink, reward_manifests_sink_server) = RewardManifest::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(DEFAULT_ROLL_TIME),
+            FileSinkCommitStrategy::Manual,
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -177,7 +177,7 @@ impl Server {
             NonRewardablePacket::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                Some(Duration::from_secs(5 * 60)),
+                FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
@@ -210,7 +210,7 @@ impl Server {
             LoraInvalidBeaconReportV1::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                Some(DEFAULT_ROLL_TIME),
+                FileSinkCommitStrategy::Manual,
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
@@ -219,7 +219,7 @@ impl Server {
             LoraInvalidWitnessReportV1::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                Some(DEFAULT_ROLL_TIME),
+                FileSinkCommitStrategy::Manual,
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
@@ -243,7 +243,7 @@ impl Server {
             LoraInvalidBeaconReportV1::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                Some(Duration::from_secs(5 * 60)),
+                FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
@@ -252,15 +252,15 @@ impl Server {
             LoraInvalidWitnessReportV1::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                Some(Duration::from_secs(5 * 60)),
+                FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(5 * 60)),
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
 
         let (runner_poc_sink, runner_poc_sink_server) = LoraPocV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(Duration::from_secs(2 * 60)),
+            FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(2 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/mobile_packet_verifier/src/daemon.rs b/mobile_packet_verifier/src/daemon.rs
@@ -8,7 +8,7 @@ use file_store::{
     file_sink::FileSinkClient,
     file_source, file_upload,
     mobile_session::DataTransferSessionIngestReport,
-    traits::FileSinkWriteExt,
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
     FileStore, FileType,
 };
 
@@ -174,7 +174,7 @@ impl Cmd {
         let (valid_sessions, valid_sessions_server) = ValidDataTransferSession::file_sink(
             store_base_path,
             file_upload.clone(),
-            None,
+            FileSinkCommitStrategy::Automatic,
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -183,15 +183,15 @@ impl Cmd {
             InvalidDataTransferIngestReportV1::file_sink(
                 store_base_path,
                 file_upload.clone(),
-                None,
+                FileSinkCommitStrategy::Manual,
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
 
         let (pending_sessions, pending_sessions_server) = PendingDataTransferSessionV1::file_sink(
             store_base_path,
             file_upload.clone(),
-            None,
+            FileSinkCommitStrategy::Manual,
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/mobile_verifier/src/boosting_oracles/data_sets.rs b/mobile_verifier/src/boosting_oracles/data_sets.rs
@@ -9,7 +9,7 @@ use chrono::{DateTime, Utc};
 use file_store::{
     file_sink::FileSinkClient,
     file_upload::FileUpload,
-    traits::{FileSinkWriteExt, TimestampDecode, TimestampEncode},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt, TimestampDecode, TimestampEncode},
     FileStore,
 };
 use futures_util::{Stream, StreamExt, TryFutureExt, TryStreamExt};
@@ -262,7 +262,7 @@ impl
             OracleBoostingReportV1::file_sink(
                 settings.store_base_path(),
                 file_upload.clone(),
-                Some(Duration::from_secs(15 * 60)),
+                FileSinkCommitStrategy::AutomaticRollTime(Duration::from_secs(15 * 60)),
                 env!("CARGO_PKG_NAME"),
             )
             .await?;

diff --git a/mobile_verifier/src/cli/server.rs b/mobile_verifier/src/cli/server.rs
@@ -15,7 +15,11 @@ use crate::{
     telemetry, Settings,
 };
 use anyhow::Result;
-use file_store::{file_upload, traits::FileSinkWriteExt, FileStore};
+use file_store::{
+    file_upload,
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
+    FileStore,
+};
 use helium_proto::services::poc_mobile::{Heartbeat, SeniorityUpdate, SpeedtestAvg};
 use mobile_config::client::{
     entity_client::EntityClient, hex_boosting_client::HexBoostingClient, AuthorizationClient,
@@ -52,7 +56,7 @@ impl Cmd {
         let (valid_heartbeats, valid_heartbeats_server) = Heartbeat::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(Duration::from_secs(15 * 60)),
+            FileSinkCommitStrategy::ManualRollTime(Duration::from_secs(15 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
@@ -61,15 +65,15 @@ impl Cmd {
         let (seniority_updates, seniority_updates_server) = SeniorityUpdate::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(Duration::from_secs(15 * 60)),
+            FileSinkCommitStrategy::ManualRollTime(Duration::from_secs(15 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;
 
         let (speedtests_avg, speedtests_avg_server) = SpeedtestAvg::file_sink(
             store_base_path,
             file_upload.clone(),
-            Some(Duration::from_secs(15 * 60)),
+            FileSinkCommitStrategy::ManualRollTime(Duration::from_secs(15 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/mobile_verifier/src/coverage.rs b/mobile_verifier/src/coverage.rs
@@ -10,7 +10,7 @@ use file_store::{
     file_sink::FileSinkClient,
     file_source,
     file_upload::FileUpload,
-    traits::{FileSinkWriteExt, TimestampEncode},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt, TimestampEncode},
     FileStore, FileType,
 };
 use futures::{
@@ -89,7 +89,7 @@ impl CoverageDaemon {
         let (valid_coverage_objs, valid_coverage_objs_server) = proto::CoverageObjectV1::file_sink(
             settings.store_base_path(),
             file_upload.clone(),
-            Some(Duration::from_secs(15 * 60)),
+            FileSinkCommitStrategy::ManualRollTime(Duration::from_secs(15 * 60)),
             env!("CARGO_PKG_NAME"),
         )
         .await?;

diff --git a/mobile_verifier/src/radio_threshold.rs b/mobile_verifier/src/radio_threshold.rs
@@ -11,7 +11,7 @@ use file_store::{
     mobile_radio_threshold::{
         RadioThresholdIngestReport, RadioThresholdReportReq, VerifiedRadioThresholdIngestReport,
     },
-    traits::{FileSinkWriteExt, DEFAULT_ROLL_TIME},
+    traits::{FileSinkCommitStrategy, FileSinkWriteExt},
     FileStore, FileType,
 };
 use futures::{StreamExt, TryStreamExt};
@@ -73,7 +73,7 @@ where
             VerifiedRadioThresholdIngestReportV1::file_sink(
                 settings.store_base_path(),
                 file_upload.clone(),
-                Some(DEFAULT_ROLL_TIME),
+                FileSinkCommitStrategy::Manual,
                 env!("CARGO_PKG_NAME"),
             )
             .await?;
@@ -82,7 +82,7 @@ where
             VerifiedInvalidatedRadioThresholdIngestReportV1::file_sink(
                 settings.store_base_path(),
                 file_upload.clone(),
-                Some(DEFAULT_ROLL_TIME),
+                FileSinkCommitStrategy::Manual,
                 env!("CARGO_PKG_NAME"),
             )
             .await?;