From dc552ac5e0fd07617ad9ce25ae46e77a0fae57b1 Mon Sep 17 00:00:00 2001
From: Ameer Ghani <inahga@divviup.org>
Date: Tue, 16 Apr 2024 16:36:49 -0400
Subject: [PATCH 1/2] Add integration test for time-bucketed fixed size

---
 Cargo.lock                                    | 18 +++++-----
 .../tests/integration/in_cluster.rs           | 35 ++++++++++++++++++-
 2 files changed, 43 insertions(+), 10 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index a5dc592fa..fb0b2b2f2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1221,13 +1221,13 @@ dependencies = [
 
 [[package]]
 name = "divviup-client"
-version = "0.2.0"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f61f3b8ea9d2af56a3e7821f5eb07930688c97be58c431e4a6920d27b3893530"
+checksum = "ecacbabd66def525c90cb0e9c0d6665d59c10ff22d87bf3f3d58d17329e4986c"
 dependencies = [
  "base64 0.22.0",
  "email_address",
- "janus_messages 0.7.2",
+ "janus_messages 0.7.5",
  "log",
  "pad-adapter",
  "serde",
@@ -2630,9 +2630,9 @@ dependencies = [
 
 [[package]]
 name = "janus_messages"
-version = "0.7.2"
+version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "425b77ad31e0614c16041aa44c4014ebd4c67544d01135cece4f51c33b4adef2"
+checksum = "bd629c85e8ff09303b2bfccb15e7ae07adb716a998d9124b312f988afbd8b5b5"
 dependencies = [
  "anyhow",
  "base64 0.22.0",
@@ -5062,9 +5062,9 @@ dependencies = [
 
 [[package]]
 name = "time"
-version = "0.3.34"
+version = "0.3.36"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8248b6521bb14bc45b4067159b9b6ad792e2d6d754d6c41fb50e29fefe38749"
+checksum = "5dfd88e563464686c916c7e46e623e520ddc6d79fa6641390f2e3fa86e83e885"
 dependencies = [
  "deranged",
  "itoa",
@@ -5083,9 +5083,9 @@ checksum = "ef927ca75afb808a4d64dd374f00a2adf8d0fcff8e7b184af886c3c87ec4a3f3"
 
 [[package]]
 name = "time-macros"
-version = "0.2.17"
+version = "0.2.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ba3a3ef41e6672a2f0f001392bb5dcd3ff0a9992d618ca761a11c3121547774"
+checksum = "3f252a68540fde3a3877aeea552b832b40ab9a69e318efd078774a01ddee1ccf"
 dependencies = [
  "num-conv",
  "time-core",
diff --git a/integration_tests/tests/integration/in_cluster.rs b/integration_tests/tests/integration/in_cluster.rs
index 1ff8f8fe0..171a336b7 100644
--- a/integration_tests/tests/integration/in_cluster.rs
+++ b/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;

From 3f5e0ef2b5f972f55c798e4aca941cc2ec3e5b8f Mon Sep 17 00:00:00 2001
From: Ameer Ghani <inahga@divviup.org>
Date: Thu, 11 Apr 2024 09:25:55 -0400
Subject: [PATCH 2/2] Advertise support for time bucketed fixed size (#3001)

---
 aggregator_api/src/routes.rs | 4 +---
 aggregator_api/src/tests.rs  | 2 +-
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/aggregator_api/src/routes.rs b/aggregator_api/src/routes.rs
index dd2af2c1c..f6abfeb42 100644
--- a/aggregator_api/src/routes.rs
+++ b/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"],
     })
 }
 
diff --git a/aggregator_api/src/tests.rs b/aggregator_api/src/tests.rs
index fcf578bd7..c511006a3 100644
--- a/aggregator_api/src/tests.rs
+++ b/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"]}"#,
         )
     );
 }