Repository: cloudflare/daphne
Pull request #187 (2023-01-03 22:09:18)

Time-based GC.
This is a first cut at time-based GC.  Processed report ids are chunked into epochs, we reject things that are too old or too far in the future, and we delete the chunk once it is impossible for any replay of the report id in that epoch.


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 6a4434c..1c8013f 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -312,31 +312,16 @@ impl AsRef<str> for DapVersion {
 #[derive(Clone, Deserialize, Serialize)]
 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:
+    /// which an Aggregator guarantees storage of reports and/or report metadata.
     ///
-    /// 1. Once an epoch has elapsed, all reports and metadata corresponding to the epoch may be
-    ///    deleted in order to reduce storage costs. In addition, all reports for this epoch will
-    ///    be rejected.
-    /// 2. Reports pertaining to epochs that have not yet begun may be rejected.
-    ///
-    /// Let `now` be the current time truncated by the report storage epoch duration. All epochs
-    /// preceeding the previous are subject to deletion; for all epochs following the next, reports
-    /// will be rejected:
-    ///
-    /// ```text
-    ///     now-2 <- reports rejected, data subject to deletion
-    ///     now-1 <- start of previous epoch
-    ///     now   <- start of current epoch
-    ///     now+1 <- start of next epoch
-    ///     now+2 <- reports rejected
-    /// ```
-    ///
-    /// Thus, storage is only guaranteed for the previous epoch, the current epoch, and the next
-    /// epoch.
-    //
-    // TODO(issue #100) Implement the rejection mechanism described here.
+    /// 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_worker/src/config.rs b/daphne_worker/src/config.rs
index 768bcda..f88f4c1 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -154,6 +154,7 @@ impl<D> DaphneWorkerConfig<D> {
         let deployment = if let Ok(deployment) = ctx.var("DAP_DEPLOYMENT") {
             match deployment.to_string().as_str() {
                 "prod" => DaphneWorkerDeployment::Prod,
+                "dev" => DaphneWorkerDeployment::Dev,
                 s => {
                     return Err(Error::RustError(format!(
                         "Invalid value for DAP_DEPLOYMENT: {}",
@@ -667,6 +668,14 @@ impl<D> DaphneWorkerConfig<D> {
             sender_auth,
         })
     }
+
+    pub(crate) fn least_valid_report_time(&self, now: u64) -> u64 {
+        now.saturating_sub(self.global_config.report_storage_epoch_duration)
+    }
+
+    pub(crate) fn greatest_valid_report_time(&self, now: u64) -> u64 {
+        now.saturating_add(self.global_config.report_storage_max_future_time_skew)
+    }
 }
 
 /// RwLockReadGuard'ed object, used to catch items fetched from KV.
@@ -716,4 +725,8 @@ pub(crate) enum DaphneWorkerDeployment {
     /// Daphne-Worker is running in a production environment. No behavior overrides are applied.
     #[default]
     Prod,
+    /// Daphne-Worker is running in a development environment. Any durable objects that are created
+    /// will be registered by the garbage collector so that they can be deleted manually using the
+    /// internal test API.
+    Dev,
 }
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index 8c2f2f3..95fad9a 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -545,9 +545,13 @@ where
         let agg_store_responses: Vec<bool> =
             try_join_all(agg_store_requests).await.map_err(dap_err)?;
 
-        // Decide which reports to reject early. A report will be rejected if has been processed
-        // but not collected or if it has not been proceessed but pertains to a batch that was
-        // previously collected.
+        // Decide which reports to reject early. A report will be rejected here if, for example,
+        // it has been processed but not collected, or if it has not been proceessed but pertains
+        // to a batch that was previously collected, or if it is not within time bounds specified
+        // by the configuration.
+        let current_time = self.get_current_time();
+        let min_time = self.least_valid_report_time(current_time);
+        let max_time = self.greatest_valid_report_time(current_time);
         let mut early_fails = HashMap::new();
         for (bucket, collected) in agg_store_request_bucket
             .iter()
@@ -559,6 +563,10 @@ where
                     early_fails.insert(metadata.id.clone(), TransitionFailure::ReportReplayed);
                 } else if !processed && collected {
                     early_fails.insert(metadata.id.clone(), TransitionFailure::BatchCollected);
+                } else if metadata.time < min_time {
+                    early_fails.insert(metadata.id.clone(), TransitionFailure::ReportDropped);
+                } else if metadata.time > max_time {
+                    early_fails.insert(metadata.id.clone(), TransitionFailure::ReportTooEarly);
                 }
             }
         }
@@ -637,7 +645,7 @@ where
                 // NOTE This check for report replay is not definitive. It's possible for two
                 // reports with the same ID to appear in two different ReportsPending instances.
                 // The definitive check is performed by DapAggregator::check_early_reject(), which
-                // tracks all repoort IDs consumed for the task in ReportsProcessed. This check
+                // tracks all report IDs consumed for the task in ReportsProcessed. This check
                 // would be too expensive to do during the upload sub-protocol.
                 Err(DapError::Transition(TransitionFailure::ReportReplayed))
             }
diff --git a/daphne_worker/src/durable/helper_state_store.rs b/daphne_worker/src/durable/helper_state_store.rs
index 4ae8195..4031b7d 100644
--- a/daphne_worker/src/durable/helper_state_store.rs
+++ b/daphne_worker/src/durable/helper_state_store.rs
@@ -1,10 +1,7 @@
 // Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use crate::{
-    durable::{state_get, state_set_if_not_exists},
-    int_err,
-};
+use crate::{durable::state_get, int_err};
 use daphne::{messages::Id, DapVersion};
 use std::time::Duration;
 use worker::*;
@@ -37,7 +34,7 @@ pub(crate) const DURABLE_HELPER_STATE_GET: &str = "/internal/do/helper_state/get
 pub struct HelperStateStore {
     state: State,
     env: Env,
-    touched: bool,
+    alarmed: bool,
 }
 
 #[durable_object]
@@ -46,27 +43,19 @@ impl DurableObject for HelperStateStore {
         Self {
             state,
             env,
-            touched: false,
+            alarmed: false,
         }
     }
 
     async fn fetch(&mut self, mut req: Request) -> Result<Response> {
         // Ensure this DO instance is garbage collected eventually.
-        if !self.touched
-            && !state_set_if_not_exists::<bool>(&self.state, "touched", &true)
-                .await?
-                .unwrap_or(false)
-        {
-            let secs: u64 = self
-                .env
-                .var("DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS")?
-                .to_string()
-                .parse()
-                .map_err(int_err)?;
-            let scheduled_time = Duration::from_secs(secs);
-            self.state.storage().set_alarm(scheduled_time).await?;
-            self.touched = true;
-        }
+        let secs: u64 = self
+            .env
+            .var("DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS")?
+            .to_string()
+            .parse()
+            .map_err(int_err)?;
+        ensure_alarmed!(self, Duration::from_secs(secs));
 
         match (req.path().as_ref(), req.method()) {
             // Store the Helper's state.
@@ -110,7 +99,7 @@ impl DurableObject for HelperStateStore {
 
     async fn alarm(&mut self) -> Result<Response> {
         self.state.storage().delete_all().await?;
-        self.touched = false;
+        self.alarmed = false;
         console_debug!(
             "HelperStateStore: deleted instance {}",
             self.state.id().to_string()
diff --git a/daphne_worker/src/durable/mod.rs b/daphne_worker/src/durable/mod.rs
index 61f721d..8b7744e 100644
--- a/daphne_worker/src/durable/mod.rs
+++ b/daphne_worker/src/durable/mod.rs
@@ -109,26 +109,43 @@ macro_rules! ensure_garbage_collected {
             $object.touched = false;
             return Response::from_json(&());
         } else if !$object.touched {
-            let touched: bool =
-                crate::durable::state_set_if_not_exists(&$object.state, "touched", &true)
-                    .await?
-                    .unwrap_or(false);
-            if !touched {
-                let durable = crate::durable::DurableConnector::new(&$object.env);
-                durable
-                    .post(
-                        crate::durable::BINDING_DAP_GARBAGE_COLLECTOR,
-                        crate::durable::garbage_collector::DURABLE_GARBAGE_COLLECTOR_PUT,
-                        "garbage_collector".to_string(),
-                        &crate::durable::DurableReference {
-                            binding: $binding.to_string(),
-                            id_hex: $id,
-                            task_id: None,
-                        },
-                    )
-                    .await?;
-                $object.touched = true;
+            // The GarbageCollector should only be used when running tests. In production, the DO->DO
+            // communication overhead adds unacceptable latency, and there's no need to do the
+            // bulk deletes of state that test suites require.
+            let deployment = $object.env.var("DAP_DEPLOYMENT")?.to_string();
+            if deployment != "prod" {
+                let touched: bool =
+                    crate::durable::state_set_if_not_exists(&$object.state, "touched", &true)
+                        .await?
+                        .unwrap_or(false);
+                if !touched {
+                    let durable = crate::durable::DurableConnector::new(&$object.env);
+                    durable
+                        .post(
+                            crate::durable::BINDING_DAP_GARBAGE_COLLECTOR,
+                            crate::durable::garbage_collector::DURABLE_GARBAGE_COLLECTOR_PUT,
+                            "garbage_collector".to_string(),
+                            &crate::durable::DurableReference {
+                                binding: $binding.to_string(),
+                                id_hex: $id,
+                                task_id: None,
+                            },
+                        )
+                        .await?;
+                }
             }
+            $object.touched = true;
+        }
+    }};
+}
+
+macro_rules! ensure_alarmed {
+    ($object:expr, $lifetime:expr) => {{
+        if !$object.alarmed {
+            if $object.state.storage().get_alarm().await?.is_none() {
+                $object.state.storage().set_alarm($lifetime).await?;
+            }
+            $object.alarmed = true;
         }
     }};
 }
diff --git a/daphne_worker/src/durable/reports_processed.rs b/daphne_worker/src/durable/reports_processed.rs
index f9c5dd1..cdffc0e 100644
--- a/daphne_worker/src/durable/reports_processed.rs
+++ b/daphne_worker/src/durable/reports_processed.rs
@@ -5,7 +5,9 @@ use crate::{
     durable::{state_set_if_not_exists, BINDING_DAP_REPORTS_PROCESSED},
     int_err,
 };
+use daphne::DapGlobalConfig;
 use futures::future::try_join_all;
+use std::time::Duration;
 use worker::*;
 
 pub(crate) const DURABLE_REPORTS_PROCESSED_MARK_AGGREGATED: &str =
@@ -30,6 +32,7 @@ pub struct ReportsProcessed {
     state: State,
     env: Env,
     touched: bool,
+    alarmed: bool,
 }
 
 impl ReportsProcessed {
@@ -54,6 +57,7 @@ impl DurableObject for ReportsProcessed {
             state,
             env,
             touched: false,
+            alarmed: false,
         }
     }
 
@@ -61,6 +65,25 @@ impl DurableObject for ReportsProcessed {
         let id_hex = self.state.id().to_string();
         ensure_garbage_collected!(req, self, id_hex.clone(), BINDING_DAP_REPORTS_PROCESSED);
 
+        let global_config: DapGlobalConfig = serde_json::from_str(
+            self.env.var("DAP_GLOBAL_CONFIG")?.to_string().as_ref(),
+        )
+        .map_err(|e| Error::RustError(format!("Failed to parse DAP_GLOBAL_CONFIG: {}", e)))?;
+        let safety_interval: u64 = self
+            .env
+            .var("DAP_PROCESSED_ALARM_SAFETY_INTERVAL")?
+            .to_string()
+            .parse()
+            .map_err(int_err)?;
+        ensure_alarmed!(
+            self,
+            Duration::from_secs(
+                global_config
+                    .report_storage_epoch_duration
+                    .saturating_add(safety_interval)
+            )
+        );
+
         match (req.path().as_ref(), req.method()) {
             // Mark a set of reports as aggregated. Return the set of report IDs that already
             // exist.
@@ -86,4 +109,11 @@ impl DurableObject for ReportsProcessed {
             ))),
         }
     }
+
+    async fn alarm(&mut self) -> Result<Response> {
+        self.state.storage().delete_all().await?;
+        self.alarmed = false;
+        self.touched = false;
+        Response::from_json(&())
+    }
 }


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index 0aba114..6fba57f 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -69,7 +69,8 @@ impl Test {
         // Global config. In a real deployment, the Leader and Helper may make different choices
         // here.
         let global_config = DapGlobalConfig {
-            report_storage_epoch_duration: 604800, // one week
+            report_storage_epoch_duration: 604800,    // one week
+            report_storage_max_future_time_skew: 300, // 5 minutes
             max_batch_duration: 360000,
             min_batch_interval_start: 259200,
             max_batch_interval_end: 259200,
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index 3ebc331..1e851ec 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -125,6 +125,7 @@ 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,
@@ -248,7 +249,8 @@ impl TestRunner {
     }
 
     pub fn batch_interval(&self) -> Interval {
-        let start = self.now - (self.now % self.task_config.time_precision);
+        let epoch_start = self.now - (self.now % self.global_config.report_storage_epoch_duration);
+        let start = epoch_start - (epoch_start % self.task_config.time_precision);
         Interval {
             start,
             duration: self.task_config.time_precision * 2,
diff --git a/daphne_worker_test/wrangler.toml b/daphne_worker_test/wrangler.toml
index b106cee..8b2ca97 100644
--- a/daphne_worker_test/wrangler.toml
+++ b/daphne_worker_test/wrangler.toml
@@ -28,6 +28,7 @@ DAP_REPORT_SHARD_KEY = "61cd9685547370cfea76c2eb8d156ad9" # SECRET
 DAP_REPORT_SHARD_COUNT = "2"
 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,
@@ -35,6 +36,8 @@ DAP_GLOBAL_CONFIG = """{
      "allow_taskprov": true,
      "taskprov_version": "v02"
 }"""
+DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
+DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
   "kem_id": "p256_hkdf_sha256",
@@ -86,6 +89,7 @@ DAP_REPORT_SHARD_COUNT = "2"
 DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = "10"
 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,
@@ -93,6 +97,8 @@ DAP_GLOBAL_CONFIG = """{
   "allow_taskprov": true,
   "taskprov_version": "v02"
 }"""
+DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
+DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
   "kem_id": "p256_hkdf_sha256",