Simulation test improvements (#3362)

* Shrink configuration when shrinking inputs

* Add quickcheck test focusing on key rotator

* Add another regression test

* Add assertion that min agg job size cannot be zero

* Add validation of durations in task constructor

* Upload multiple reports per Upload operation

* Add operation to upload without rounding timestamp

* Add operation to upload an invalid report

* Add timing of setup and operations

* Avoid thrashing ephemeral database container

* Reduce bias toward upload operations

* Discard inputs with no operations

* Change default quickcheck input size range

* Implement shrinking for upload operations

* Fix name in Debug implementation

* Log error message on timeout

* Configure collector backoff settings

* Un-ignore regression test

* Combine operations where possible when shrinking

* Add log statements with inputs

* Treat all server errors as failures

aggregator/src/aggregator/aggregation_job_creator.rs

@@ -101,6 +101,10 @@ impl<C: Clock + 'static> AggregationJobCreator<C> {
         max_aggregation_job_size: usize,
         aggregation_job_creation_report_window: usize,
     ) -> AggregationJobCreator<C> {
+        assert!(
+            min_aggregation_job_size > 0,
+            "invalid configuration: min_aggregation_job_size cannot be zero"
+        );
         assert!(
             max_aggregation_job_size > 0,
             "invalid configuration: max_aggregation_job_size cannot be zero"

aggregator_core/src/datastore/models.rs

@@ -2333,6 +2333,38 @@ impl TaskUploadCounter {
     pub fn increment_task_expired(&mut self) {
         self.task_expired += 1
     }
+
+    pub fn interval_collected(&self) -> u64 {
+        self.interval_collected
+    }
+
+    pub fn report_decode_failure(&self) -> u64 {
+        self.report_decode_failure
+    }
+
+    pub fn report_decrypt_failure(&self) -> u64 {
+        self.report_decrypt_failure
+    }
+
+    pub fn report_expired(&self) -> u64 {
+        self.report_expired
+    }
+
+    pub fn report_outdated_key(&self) -> u64 {
+        self.report_outdated_key
+    }
+
+    pub fn report_success(&self) -> u64 {
+        self.report_success
+    }
+
+    pub fn report_too_early(&self) -> u64 {
+        self.report_too_early
+    }
+
+    pub fn task_expired(&self) -> u64 {
+        self.task_expired
+    }
 }
 
 /// Per-task counts of aggregated reports.

aggregator_core/src/datastore/test_util.rs

@@ -39,15 +39,15 @@ use tracing::trace;
 
 use super::SUPPORTED_SCHEMA_VERSIONS;
 
-struct EphemeralDatabase {
+pub struct EphemeralDatabase {
     db_thread: Option<JoinHandle<()>>,
     db_thread_shutdown: Option<Sender<()>>,
     port_number: u16,
     cleanup_tx: mpsc::Sender<String>,
 }
 
 impl EphemeralDatabase {
-    async fn shared() -> Arc<Self> {
+    pub async fn shared() -> Arc<Self> {
         static EPHEMERAL_DATABASE: Mutex<Weak<EphemeralDatabase>> = Mutex::const_new(Weak::new());
 
         let mut g = EPHEMERAL_DATABASE.lock().await;

aggregator_core/src/task.rs

@@ -161,12 +161,15 @@ impl CommonTaskParameters {
 
         if let QueryType::FixedSize {
             max_batch_size: Some(max_batch_size),
-            ..
+            batch_time_window_size: Some(batch_time_window_size),
         } = query_type
         {
             if max_batch_size < min_batch_size {
                 return Err(Error::InvalidParameter("max_batch_size"));
             }
+            if batch_time_window_size.as_seconds() == 0 {
+                return Err(Error::InvalidParameter("batch_time_window_size is zero"));
+            }
         }
 
         // These fields are stored as 64-bit signed integers in the database but are held in
@@ -183,6 +186,10 @@ impl CommonTaskParameters {
                 .map_err(|_| Error::InvalidParameter("task_expiration out of range"))?;
         }
 
+        if time_precision.as_seconds() == 0 {
+            return Err(Error::InvalidParameter("time_precision is zero"));
+        }
+
         Ok(Self {
             task_id,
             query_type,