Time-bucketed fixed size support (release/0.6 backport)

aggregator_api/src/routes.rs

@@ -48,9 +48,7 @@ pub(super) async fn get_config(
             SupportedQueryType::TimeInterval,
             SupportedQueryType::FixedSize,
         ],
-        // Unconditionally indicate to divviup-api that we support collector auth token hashes and
-        // upload metrics.
-        features: &["TokenHash", "UploadMetrics"],
+        features: &["TokenHash", "UploadMetrics", "TimeBucketedFixedSize"],
     })
 }
 

aggregator_api/src/tests.rs

@@ -90,7 +90,7 @@ async fn get_config() {
             r#"{"protocol":"DAP-07","dap_url":"https://dap.url/","role":"Either","vdafs":"#,
             r#"["Prio3Count","Prio3Sum","Prio3Histogram","Prio3CountVec","Prio3SumVec"],"#,
             r#""query_types":["TimeInterval","FixedSize"],"#,
-            r#""features":["TokenHash","UploadMetrics"]}"#,
+            r#""features":["TokenHash","UploadMetrics","TimeBucketedFixedSize"]}"#,
         )
     );
 }

integration_tests/tests/integration/in_cluster.rs

@@ -17,10 +17,11 @@ use janus_core::{
         install_test_trace_subscriber,
         kubernetes::{Cluster, PortForward},
     },
+    time::DurationExt,
     vdaf::VdafInstance,
 };
 use janus_integration_tests::{client::ClientBackend, TaskParameters};
-use janus_messages::TaskId;
+use janus_messages::{Duration as JanusDuration, TaskId};
 use std::{env, str::FromStr, time::Duration};
 use trillium_rustls::RustlsConfig;
 use trillium_tokio::ClientConfig;
@@ -430,6 +431,13 @@ impl InClusterJanusPair {
                 QueryType::TimeInterval => None,
                 QueryType::FixedSize { max_batch_size, .. } => Some(*max_batch_size),
             },
+            batch_time_window_size_seconds: match task.query_type() {
+                QueryType::TimeInterval => None,
+                QueryType::FixedSize {
+                    batch_time_window_size,
+                    ..
+                } => batch_time_window_size.map(|window| window.as_seconds()),
+            },
             time_precision_seconds: task.time_precision().as_seconds(),
             collector_credential_id,
         };
@@ -561,6 +569,31 @@ async fn in_cluster_fixed_size() {
     .await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn in_cluster_time_bucketed_fixed_size() {
+    install_test_trace_subscriber();
+    initialize_rustls();
+
+    // Start port forwards and set up task.
+    let janus_pair = InClusterJanusPair::new(
+        VdafInstance::Prio3Count,
+        QueryType::FixedSize {
+            max_batch_size: Some(110),
+            batch_time_window_size: Some(JanusDuration::from_hours(8).unwrap()),
+        },
+    )
+    .await;
+
+    // Run the behavioral test.
+    submit_measurements_and_verify_aggregate(
+        "in_cluster_time_bucketed_fixed_size",
+        &janus_pair.task_parameters,
+        (janus_pair.leader.port(), janus_pair.helper.port()),
+        &ClientBackend::InProcess,
+    )
+    .await;
+}
+
 #[cfg(feature = "in-cluster-rate-limits")]
 mod rate_limits {
     use super::InClusterJanusPair;