Repository: cloudflare/daphne
Pull request #457 (2023-12-06 17:12:31)

Remove storage related settings from global config
These fields are unused by the `daphne` crate, hence they shouldn't be required
by it.



Patch:
diff --git a/daphne/Cargo.toml b/daphne/Cargo.toml
index c7598c9..7fd35bc 100644
--- a/daphne/Cargo.toml
+++ b/daphne/Cargo.toml
@@ -39,6 +39,7 @@ url.workspace = true
 
 [dev-dependencies]
 assert_matches.workspace = true
+deepsize.workspace = true
 futures.workspace = true
 matchit.workspace = true
 paste.workspace = true
diff --git a/daphne/dapf/src/bin/dapf.rs b/daphne/dapf/src/bin/dapf.rs
index b0eca86..c958c49 100644
--- a/daphne/dapf/src/bin/dapf.rs
+++ b/daphne/dapf/src/bin/dapf.rs
@@ -7,7 +7,7 @@ use daphne::{
     constants::DapMediaType,
     error::aborts::ProblemDetails,
     hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
-    messages::{BatchSelector, Collection, CollectionReq, Query, TaskId},
+    messages::{Base64Encode, BatchSelector, Collection, CollectionReq, Query, TaskId},
     DapMeasurement, DapVersion, VdafConfig,
 };
 use prio::codec::{Decode, ParameterizedDecode, ParameterizedEncode};
diff --git a/daphne/src/error/aborts.rs b/daphne/src/error/aborts.rs
index c5421a7..022b514 100644
--- a/daphne/src/error/aborts.rs
+++ b/daphne/src/error/aborts.rs
@@ -5,7 +5,7 @@
 
 use crate::{
     fatal_error,
-    messages::{BatchSelector, TaskId, TransitionFailure},
+    messages::{Base64Encode, BatchSelector, TaskId, TransitionFailure},
     DapError, DapMediaType, DapRequest, DapVersion,
 };
 use hex::FromHexError;
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 0c4ec7f..a363871 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -83,7 +83,7 @@ use crate::{
 use constants::DapMediaType;
 pub use error::DapError;
 use hpke::{HpkeConfig, HpkeKemId};
-use messages::encode_base64url;
+use messages::{encode_base64url, Base64Encode};
 use prio::{
     codec::{Decode, Encode, ParameterizedDecode, ParameterizedEncode},
     vdaf::Aggregatable as AggregatableTrait,
@@ -141,17 +141,6 @@ impl std::fmt::Display for DapVersion {
 #[derive(Clone, Deserialize, Serialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub struct DapGlobalConfig {
-    /// The report storage epoch duration. This value is used to control the period of time for
-    /// which an Aggregator guarantees storage of reports and/or report metadata.
-    ///
-    /// A report will be accepted if its timestamp is no more than the specified number of seconds
-    /// before the current time.
-    pub report_storage_epoch_duration: Duration,
-
-    /// The report storage maximum future time skew. Reports with timestamps greater than the
-    /// current time plus this value will be rejected.
-    pub report_storage_max_future_time_skew: Duration,
-
     /// Maximum interval duration permitted in CollectReq.
     /// Prevents Collectors from requesting wide range or reports.
     pub max_batch_duration: Duration,
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index dfba589..3f8ca92 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -34,6 +34,16 @@ const FIXED_SIZE_QUERY_TYPE_CURRENT_BATCH: u8 = 0x01;
 // Known extension types.
 const EXTENSION_TASKPROV: u16 = 0xff00;
 
+pub trait Base64Encode {
+    /// Encode to URL-safe base64.
+    fn to_base64url(&self) -> String;
+
+    /// Decode from URL-safe, base64.
+    fn try_from_base64url<T: AsRef<str>>(id_base64url: T) -> Option<Self>
+    where
+        Self: Sized;
+}
+
 // Serde doesn't support derivations from const generics properly, so we have to use a macro.
 macro_rules! id_struct {
     ($sname:ident, $len:expr, $doc:expr) => {
@@ -45,18 +55,20 @@ macro_rules! id_struct {
         pub struct $sname(#[serde(with = "hex")] pub [u8; $len]);
 
         impl $sname {
-            /// Return the URL-safe, base64 encoding of the ID.
-            pub fn to_base64url(&self) -> String {
-                encode_base64url(self.0)
-            }
-
             /// Return the ID encoded as a hex string.
             pub fn to_hex(&self) -> String {
                 hex::encode(self.0)
             }
+        }
+
+        impl $crate::messages::Base64Encode for $sname {
+            /// Return the URL-safe, base64 encoding of the ID.
+            fn to_base64url(&self) -> String {
+                encode_base64url(self.0)
+            }
 
             /// Decode from URL-safe, base64.
-            pub fn try_from_base64url<T: AsRef<str>>(id_base64url: T) -> Option<Self> {
+            fn try_from_base64url<T: AsRef<str>>(id_base64url: T) -> Option<Self> {
                 Some($sname(decode_base64url(id_base64url.as_ref())?))
             }
         }
diff --git a/daphne/src/roles/leader.rs b/daphne/src/roles/leader.rs
index 84e5a1e..3225f52 100644
--- a/daphne/src/roles/leader.rs
+++ b/daphne/src/roles/leader.rs
@@ -17,8 +17,9 @@ use crate::{
     error::DapAbort,
     fatal_error,
     messages::{
-        AggregateShare, AggregateShareReq, AggregationJobResp, BatchSelector, Collection,
-        CollectionJobId, CollectionReq, Interval, PartialBatchSelector, Query, Report, TaskId,
+        AggregateShare, AggregateShareReq, AggregationJobResp, Base64Encode, BatchSelector,
+        Collection, CollectionJobId, CollectionReq, Interval, PartialBatchSelector, Query, Report,
+        TaskId,
     },
     metrics::DaphneRequestType,
     DapCollectJob, DapError, DapLeaderAggregationJobTransition, DapLeaderProcessTelemetry,
diff --git a/daphne/src/roles/mod.rs b/daphne/src/roles/mod.rs
index eb43e9d..d7bad9c 100644
--- a/daphne/src/roles/mod.rs
+++ b/daphne/src/roles/mod.rs
@@ -9,7 +9,7 @@ pub mod leader;
 
 use crate::{
     constants::DapMediaType,
-    messages::{BatchSelector, ReportMetadata, TaskId, Time},
+    messages::{Base64Encode, BatchSelector, ReportMetadata, TaskId, Time},
     taskprov, DapAbort, DapError, DapQueryConfig, DapRequest, DapTaskConfig,
 };
 use tracing::warn;
@@ -163,9 +163,10 @@ mod test {
         hpke::{HpkeDecrypter, HpkeKemId, HpkeReceiverConfig},
         messages::{
             AggregateShareReq, AggregationJobContinueReq, AggregationJobInitReq,
-            AggregationJobResp, BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq,
-            Extension, HpkeCiphertext, Interval, PartialBatchSelector, Query, Report, ReportId,
-            ReportMetadata, TaskId, Time, Transition, TransitionFailure, TransitionVar,
+            AggregationJobResp, Base64Encode, BatchId, BatchSelector, Collection, CollectionJobId,
+            CollectionReq, Extension, HpkeCiphertext, Interval, PartialBatchSelector, Query,
+            Report, ReportId, ReportMetadata, TaskId, Time, Transition, TransitionFailure,
+            TransitionVar,
         },
         test_versions,
         testing::{AggStore, MockAggregator, MockAggregatorReportSelector},
@@ -260,8 +261,6 @@ mod test {
             // Global config. In a real deployment, the Leader and Helper may make different choices
             // here.
             let global_config = DapGlobalConfig {
-                report_storage_epoch_duration: 604_800,   // one week
-                report_storage_max_future_time_skew: 300, // 5 minutes
                 max_batch_duration: 360_000,
                 min_batch_interval_start: 259_200,
                 max_batch_interval_end: 259_200,
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index 6a5c9c8..f50e172 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -14,9 +14,9 @@ use crate::{
     hpke::{HpkeConfig, HpkeDecrypter},
     messages::{
         encode_u32_bytes, AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp,
-        BatchSelector, Extension, HpkeCiphertext, PartialBatchSelector, PlaintextInputShare,
-        PrepareInit, Report, ReportId, ReportMetadata, ReportShare, TaskId, Time, Transition,
-        TransitionFailure, TransitionVar,
+        Base64Encode, BatchSelector, Extension, HpkeCiphertext, PartialBatchSelector,
+        PlaintextInputShare, PrepareInit, Report, ReportId, ReportMetadata, ReportShare, TaskId,
+        Time, Transition, TransitionFailure, TransitionVar,
     },
     metrics::DaphneMetrics,
     roles::DapReportInitializer,
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index 0938804..e4a1077 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -29,7 +29,8 @@ use daphne::{
     fatal_error,
     hpke::{HpkeConfig, HpkeReceiverConfig},
     messages::{
-        decode_base64url_vec, AggregationJobId, BatchId, CollectionJobId, ReportId, TaskId, Time,
+        decode_base64url_vec, AggregationJobId, Base64Encode, BatchId, CollectionJobId, ReportId,
+        TaskId, Time,
     },
     DapError, DapGlobalConfig, DapQueryConfig, DapRequest, DapResource, DapResponse, DapTaskConfig,
     DapVersion, Prio3Config, VdafConfig,
@@ -123,6 +124,17 @@ pub(crate) struct DaphneWorkerConfig {
 
     /// Metrics push configuration.
     metrics_push_config: Option<MetricsPushConfig>,
+
+    /// The report storage epoch duration. This value is used to control the period of time for
+    /// which an Aggregator guarantees storage of reports and/or report metadata.
+    ///
+    /// A report will be accepted if its timestamp is no more than the specified number of seconds
+    /// before the current time.
+    pub report_storage_epoch_duration: daphne::messages::Duration,
+
+    /// The report storage maximum future time skew. Reports with timestamps greater than the
+    /// current time plus this value will be rejected.
+    pub report_storage_max_future_time_skew: daphne::messages::Duration,
 }
 
 impl DaphneWorkerConfig {
@@ -155,6 +167,26 @@ impl DaphneWorkerConfig {
         )
         .map_err(|e| worker::Error::RustError(format!("Failed to parse DAP_GLOBAL_CONFIG: {e}")))?;
 
+        let report_storage_epoch_duration = env
+            .var("DAP_REPORT_STORAGE_EPOCH_DURATION")?
+            .to_string()
+            .parse()
+            .map_err(|e| {
+                worker::Error::RustError(format!(
+                    "failed to parse DAP_REPORT_STORAGE_EPOCH_DURATION: {e:?}"
+                ))
+            })?;
+
+        let report_storage_max_future_time_skew = env
+            .var("DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW")?
+            .to_string()
+            .parse()
+            .map_err(|e| {
+                worker::Error::RustError(format!(
+                    "failed to parse DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW: {e:?}"
+                ))
+            })?;
+
         let default_version: DapVersion = env
             .var("DAP_DEFAULT_VERSION")?
             .to_string()
@@ -317,6 +349,8 @@ impl DaphneWorkerConfig {
             default_version,
             helper_state_store_garbage_collect_after_secs,
             metrics_push_config,
+            report_storage_epoch_duration,
+            report_storage_max_future_time_skew,
         })
     }
 
@@ -335,7 +369,7 @@ impl DaphneWorkerConfig {
                 .try_into()
                 .unwrap(),
         ) % self.report_shard_count;
-        let epoch = report_time - (report_time % self.global.report_storage_epoch_duration);
+        let epoch = report_time - (report_time % self.report_storage_epoch_duration);
         durable_name_report_store(task_config.version, task_id_hex, epoch, shard)
     }
 }
@@ -1170,11 +1204,11 @@ impl<'srv> DaphneWorker<'srv> {
     }
 
     pub(crate) fn least_valid_report_time(&self, now: u64) -> u64 {
-        now.saturating_sub(self.config().global.report_storage_epoch_duration)
+        now.saturating_sub(self.config().report_storage_epoch_duration)
     }
 
     pub(crate) fn greatest_valid_report_time(&self, now: u64) -> u64 {
-        now.saturating_add(self.config().global.report_storage_max_future_time_skew)
+        now.saturating_add(self.config().report_storage_max_future_time_skew)
     }
 
     // Generic HTTP POST/PUT
diff --git a/daphne_worker/src/durable/leader_col_job_queue.rs b/daphne_worker/src/durable/leader_col_job_queue.rs
index d39a24c..851863d 100644
--- a/daphne_worker/src/durable/leader_col_job_queue.rs
+++ b/daphne_worker/src/durable/leader_col_job_queue.rs
@@ -12,7 +12,7 @@ use crate::{
     initialize_tracing, int_err,
 };
 use daphne::{
-    messages::{Collection, CollectionJobId, CollectionReq, TaskId},
+    messages::{Base64Encode, Collection, CollectionJobId, CollectionReq, TaskId},
     DapCollectJob, DapVersion,
 };
 use prio::codec::ParameterizedEncode;
diff --git a/daphne_worker/src/roles/leader.rs b/daphne_worker/src/roles/leader.rs
index 1661be8..dd021b8 100644
--- a/daphne_worker/src/roles/leader.rs
+++ b/daphne_worker/src/roles/leader.rs
@@ -33,8 +33,8 @@ use daphne::{
     error::DapAbort,
     fatal_error,
     messages::{
-        Collection, CollectionJobId, CollectionReq, PartialBatchSelector, Report, TaskId,
-        TransitionFailure,
+        Base64Encode, Collection, CollectionJobId, CollectionReq, PartialBatchSelector, Report,
+        TaskId, TransitionFailure,
     },
     roles::{DapAuthorizedSender, DapLeader},
     DapCollectJob, DapError, DapQueryConfig, DapRequest, DapResponse, DapTaskConfig,
diff --git a/daphne_worker/src/router/leader.rs b/daphne_worker/src/router/leader.rs
index ac3e5ae..ecff322 100644
--- a/daphne_worker/src/router/leader.rs
+++ b/daphne_worker/src/router/leader.rs
@@ -4,7 +4,7 @@
 use daphne::{
     constants::DapMediaType,
     error::DapAbort,
-    messages::{CollectionJobId, TaskId},
+    messages::{Base64Encode, CollectionJobId, TaskId},
     roles::{leader, DapLeader},
     DapCollectJob, DapRequest, DapResponse, DapVersion,
 };
diff --git a/docker/wrangler.toml b/docker/wrangler.toml
index ce27352..341b377 100644
--- a/docker/wrangler.toml
+++ b/docker/wrangler.toml
@@ -27,9 +27,9 @@ DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
 DAP_COLLECTION_JOB_ID_KEY = "b416a85d280591d6da14e5b75a7d6e31" # SECRET
 DAP_REPORT_SHARD_KEY = "61cd9685547370cfea76c2eb8d156ad9" # SECRET
 DAP_REPORT_SHARD_COUNT = "2"
+DAP_REPORT_STORAGE_EPOCH_DURATION = "604800"
+DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW = "300"
 DAP_GLOBAL_CONFIG = """{
-     "report_storage_epoch_duration": 604800,
-     "report_storage_max_future_time_skew": 300,
      "max_batch_duration": 360000,
      "min_batch_interval_start": 259200,
      "max_batch_interval_end": 259200,
@@ -105,9 +105,9 @@ DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
 DAP_REPORT_SHARD_KEY = "f79c352056982bae1737e34bdac24d63" # SECRET
 DAP_REPORT_SHARD_COUNT = "2"
 DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = "10"
+DAP_REPORT_STORAGE_EPOCH_DURATION = "604800"
+DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW = "300"
 DAP_GLOBAL_CONFIG = """{
-  "report_storage_epoch_duration": 604800,
-  "report_storage_max_future_time_skew": 300,
   "max_batch_duration": 360000,
   "min_batch_interval_start": 259200,
   "max_batch_interval_end": 259200,


Test patch:
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index e201e21..713e105 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -10,9 +10,9 @@ use crate::{
     fatal_error,
     hpke::{HpkeConfig, HpkeDecrypter, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp, BatchId,
-        BatchSelector, Collection, CollectionJobId, CollectionReq, HpkeCiphertext, Interval,
-        PartialBatchSelector, Report, ReportId, TaskId, Time, TransitionFailure,
+        AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp, Base64Encode,
+        BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq, HpkeCiphertext,
+        Interval, PartialBatchSelector, Report, ReportId, TaskId, Time, TransitionFailure,
     },
     metrics::DaphneMetrics,
     roles::{
diff --git a/daphne_worker/src/router/test_routes.rs b/daphne_worker/src/router/test_routes.rs
index ec39ced..af6dfcd 100644
--- a/daphne_worker/src/router/test_routes.rs
+++ b/daphne_worker/src/router/test_routes.rs
@@ -4,7 +4,7 @@
 use daphne::{
     error::DapAbort,
     hpke::HpkeReceiverConfig,
-    messages::{Duration, TaskId, Time},
+    messages::{Base64Encode, Duration, TaskId, Time},
     roles::leader,
 };
 use serde::Deserialize;
diff --git a/daphne_worker_test/tests/e2e/test_runner.rs b/daphne_worker_test/tests/e2e/test_runner.rs
index d5dee9d..6b75c6b 100644
--- a/daphne_worker_test/tests/e2e/test_runner.rs
+++ b/daphne_worker_test/tests/e2e/test_runner.rs
@@ -8,7 +8,8 @@ use daphne::{
     constants::DapMediaType,
     hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        encode_base64url, BatchId, CollectionJobId, Duration, HpkeConfigList, Interval, TaskId,
+        encode_base64url, Base64Encode, BatchId, CollectionJobId, Duration, HpkeConfigList,
+        Interval, TaskId,
     },
     DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
     Prio3Config, VdafConfig,
@@ -111,8 +112,6 @@ impl TestRunner {
 
         // This block needs to be kept in-sync with daphne_worker_test/wrangler.toml.
         let global_config = DapGlobalConfig {
-            report_storage_epoch_duration: 604800,
-            report_storage_max_future_time_skew: 300,
             max_batch_duration: 360000,
             min_batch_interval_start: 259200,
             max_batch_interval_end: 259200,
@@ -276,9 +275,10 @@ impl TestRunner {
     }
 
     pub fn report_interval(&self, interval: &Interval) -> Range<u64> {
+        const REPORT_STORAGE_MAX_FUTURE_TIME_SKEW: u64 = 300;
         // This is a portion of the interval which is guaranteed to be a valid report time
         // provided that the interval start time is valid.
-        interval.start..interval.start + self.global_config.report_storage_max_future_time_skew
+        interval.start..interval.start + REPORT_STORAGE_MAX_FUTURE_TIME_SKEW
     }
 
     pub async fn get_hpke_configs(
diff --git a/daphne_worker_test/wrangler.toml b/daphne_worker_test/wrangler.toml
index 6103c9f..482523f 100644
--- a/daphne_worker_test/wrangler.toml
+++ b/daphne_worker_test/wrangler.toml
@@ -27,9 +27,9 @@ DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
 DAP_COLLECTION_JOB_ID_KEY = "12da249b0e3b1a9b5936d6ff6cdd2fa09964e2e11f5450e04eef11dc5e64daf1" # SECRET
 DAP_REPORT_SHARD_KEY = "b6e5736df46d9dd7dc087cad1cd4f03c1ff399160a8eb30c5e7841aa696e737f" # SECRET
 DAP_REPORT_SHARD_COUNT = "2"
+DAP_REPORT_STORAGE_EPOCH_DURATION = "604800"
+DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW = "300"
 DAP_GLOBAL_CONFIG = """{
-     "report_storage_epoch_duration": 604800,
-     "report_storage_max_future_time_skew": 300,
      "max_batch_duration": 360000,
      "min_batch_interval_start": 259200,
      "max_batch_interval_end": 259200,
@@ -105,9 +105,9 @@ DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
 DAP_REPORT_SHARD_KEY = "896eccc5e89474dd656d490408226fdd4ce5b34d2d1349147c6fcbb1741ddb2d" # SECRET
 DAP_REPORT_SHARD_COUNT = "2"
 DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = "10"
+DAP_REPORT_STORAGE_EPOCH_DURATION = "604800"
+DAP_REPORT_STORAGE_MAX_FUTURE_TIME_SKEW = "300"
 DAP_GLOBAL_CONFIG = """{
-  "report_storage_epoch_duration": 604800,
-  "report_storage_max_future_time_skew": 300,
   "max_batch_duration": 360000,
   "min_batch_interval_start": 259200,
   "max_batch_interval_end": 259200,