diff --git a/.github/workflows/ci-build.yml b/.github/workflows/ci-build.yml index ddeeba44b..983005197 100644 --- a/.github/workflows/ci-build.yml +++ b/.github/workflows/ci-build.yml @@ -90,13 +90,13 @@ jobs: - name: Build (Docker-specific tests) # This matches the `cargo test` command invoked by `cargo xtask test-docker`, except for `--no-run`. run: cargo test --profile ci --locked --package=janus_interop_binaries --package=janus_integration_tests --features=testcontainer --no-run - #- name: Test (Docker) - # id: test-docker - # env: - # RUST_LOG: info - # JANUS_E2E_LOGS_PATH: ${{ github.workspace }}/test-logs - # DIVVIUP_TS_INTEROP_CONTAINER: ${{ steps.default-input-values.outputs.divviup_ts_interop_container }} - # run: cargo run --package xtask --profile=ci -- test-docker --profile=ci --locked + - name: Test (Docker) + id: test-docker + env: + RUST_LOG: info + JANUS_E2E_LOGS_PATH: ${{ github.workspace }}/test-logs + DIVVIUP_TS_INTEROP_CONTAINER: ${{ steps.default-input-values.outputs.divviup_ts_interop_container }} + run: cargo run --package xtask --profile=ci -- test-docker --profile=ci --locked - name: Upload container logs # Only get logs on test failure if: failure() && (steps.test.outcome == 'failure' || steps.test-docker.outcome == 'failure') diff --git a/integration_tests/tests/integration/common.rs b/integration_tests/tests/integration/common.rs index 92d36f183..9021deef9 100644 --- a/integration_tests/tests/integration/common.rs +++ b/integration_tests/tests/integration/common.rs @@ -15,7 +15,7 @@ use janus_integration_tests::{ }; use janus_messages::{ Duration, Interval, Query, Time, TimePrecision, - batch_mode::{self, LeaderSelected}, + batch_mode::{self, LeaderSelected, LeaderSelectedQueryConfig}, problem_type::DapProblemType, }; use prio::{ @@ -325,13 +325,12 @@ where BatchMode::LeaderSelected { .. } => { let mut requests = 0; + let idempotency_key: LeaderSelectedQueryConfig = random(); let collection = loop { requests += 1; let collection_res = collect_generic::<_, LeaderSelected>( &collector, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(idempotency_key.clone()), aggregation_parameter, ) .await; diff --git a/integration_tests/tests/integration/simulation/arbitrary.rs b/integration_tests/tests/integration/simulation/arbitrary.rs index 31638d476..60bffcfc5 100644 --- a/integration_tests/tests/integration/simulation/arbitrary.rs +++ b/integration_tests/tests/integration/simulation/arbitrary.rs @@ -9,13 +9,15 @@ use std::cmp::max; use chrono::TimeDelta; use janus_aggregator_core::task::AggregationMode; use janus_core::time::{DateTimeExt, TimeDeltaExt, TimeExt}; -use janus_messages::{CollectionJobId, Duration, Interval, Time, TimePrecision}; +use janus_messages::{ + Duration, Interval, Time, TimePrecision, batch_mode::LeaderSelectedQueryConfig, +}; use quickcheck::{Arbitrary, Gen, empty_shrinker}; use rand::random; use crate::simulation::{ START_TIME, - model::{Config, Input, Op}, + model::{Config, Input, Op, Query}, }; impl Arbitrary for Config { @@ -86,8 +88,8 @@ pub(super) struct KeyRotatorInput(pub(super) Input); struct Context { current_time: Time, time_precision: TimePrecision, - started_collection_job_ids: Vec, - polled_collection_job_ids: Vec, + started_collection_job_queries: Vec, + polled_collection_job_queries: Vec, } impl Context { @@ -95,8 +97,8 @@ impl Context { Self { current_time: START_TIME.to_time(&config.time_precision), time_precision: config.time_precision, - started_collection_job_ids: Vec::new(), - polled_collection_job_ids: Vec::new(), + started_collection_job_queries: Vec::new(), + polled_collection_job_queries: Vec::new(), } } @@ -108,14 +110,9 @@ impl Context { .add_timedelta(amount, &self.time_precision) .unwrap() } - Op::CollectorStart { - collection_job_id, - query: _, - } => self.started_collection_job_ids.push(*collection_job_id), - Op::CollectorPoll { collection_job_id } - if !self.polled_collection_job_ids.contains(collection_job_id) => - { - self.polled_collection_job_ids.push(*collection_job_id); + Op::CollectorStart { query } => self.started_collection_job_queries.push(query.clone()), + Op::CollectorPoll { query } if !self.polled_collection_job_queries.contains(query) => { + self.polled_collection_job_queries.push(query.clone()); } _ => {} } @@ -187,14 +184,8 @@ fn shrink_op(op: &Op) -> Box> { | Op::CollectionJobDriver | Op::CollectionJobDriverRequestError | Op::CollectionJobDriverResponseError - | Op::CollectorStart { - collection_job_id: _, - query: _, - } - | Op::CollectorPoll { - collection_job_id: _, - } => empty_shrinker(), - + | Op::CollectorStart { .. } + | Op::CollectorPoll { .. } => empty_shrinker(), Op::Upload { report_time: _, count: 0 | 1, @@ -297,21 +288,30 @@ fn arbitrary_collector_start_op_time_interval(g: &mut Gen, context: &Context) -> ]) .unwrap(); Op::CollectorStart { - collection_job_id: random(), - query: super::model::Query::TimeInterval( - Interval::new(start, duration_fn(g, context)).unwrap(), - ), + query: Query::TimeInterval(Interval::new(start, duration_fn(g, context)).unwrap()), + } +} + +/// Generate a collect poll operation for time interval batch mode. +fn arbitrary_collector_poll_op_time_interval(g: &mut Gen, context: &Context) -> Op { + Op::CollectorPoll { + query: g + .choose(&context.started_collection_job_queries) + .cloned() + .unwrap_or_else(|| { + Query::TimeInterval(Interval::minimal(Time::from_time_precision_units(0)).unwrap()) + }), } } -/// Generate a collect poll operation. -fn arbitrary_collector_poll_op(g: &mut Gen, context: &Context) -> Op { +/// Generate a collect poll operation for leader selected batch mode. +fn arbitrary_collector_poll_op_leader_selected(g: &mut Gen, context: &Context) -> Op { Op::CollectorPoll { - collection_job_id: g - .choose(&context.started_collection_job_ids) - .copied() + query: g + .choose(&context.started_collection_job_queries) + .cloned() .unwrap_or_else(|| { - CollectionJobId::try_from([0u8; CollectionJobId::LEN].as_slice()).unwrap() + Query::LeaderSelected(LeaderSelectedQueryConfig::new(vec![0u8; 8]).unwrap()) }), } } @@ -438,7 +438,7 @@ fn arbitrary_op_time_interval(g: &mut Gen, context: &Context, choices: &[OpKind] OpKind::CollectionJobDriverRequestError => Op::CollectionJobDriverRequestError, OpKind::CollectionJobDriverResponseError => Op::CollectionJobDriverResponseError, OpKind::CollectorStart => arbitrary_collector_start_op_time_interval(g, context), - OpKind::CollectorPoll => arbitrary_collector_poll_op(g, context), + OpKind::CollectorPoll => arbitrary_collector_poll_op_time_interval(g, context), } } @@ -491,10 +491,9 @@ fn arbitrary_op_leader_selected(g: &mut Gen, context: &Context, choices: &[OpKin OpKind::CollectionJobDriverRequestError => Op::CollectionJobDriverRequestError, OpKind::CollectionJobDriverResponseError => Op::CollectionJobDriverResponseError, OpKind::CollectorStart => Op::CollectorStart { - collection_job_id: random(), - query: super::model::Query::LeaderSelected, + query: Query::LeaderSelected(random()), }, - OpKind::CollectorPoll => arbitrary_collector_poll_op(g, context), + OpKind::CollectorPoll => arbitrary_collector_poll_op_leader_selected(g, context), } } diff --git a/integration_tests/tests/integration/simulation/model.rs b/integration_tests/tests/integration/simulation/model.rs index e1a0baa44..f90dae95a 100644 --- a/integration_tests/tests/integration/simulation/model.rs +++ b/integration_tests/tests/integration/simulation/model.rs @@ -1,6 +1,8 @@ use chrono::TimeDelta; use janus_aggregator_core::task::AggregationMode; -use janus_messages::{CollectionJobId, Duration, Interval, Time, TimePrecision}; +use janus_messages::{ + Duration, Interval, Time, TimePrecision, batch_mode::LeaderSelectedQueryConfig, +}; #[derive(Debug, Clone)] pub(super) struct Input { @@ -121,21 +123,22 @@ pub(super) enum Op { /// Forward all requests, but drop the responses, and return some sort of error. CollectionJobDriverResponseError, - /// The collector sends a collection request to the leader. It remembers the collection job ID. - CollectorStart { - collection_job_id: CollectionJobId, - query: Query, - }, + /// The collector sends a collection request to the leader. + /// + /// It remembers a mapping of queries to the collection job IDs. + CollectorStart { query: Query }, /// The collector sends a request to the leader to poll an existing collection job. - CollectorPoll { collection_job_id: CollectionJobId }, + /// + /// The collection job matching the query is looked up in the collector state. + CollectorPoll { query: Query }, } /// Representation of a DAP query used in a collection job. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub(super) enum Query { /// A time interval query, parameterized with a batch interval. TimeInterval(Interval), /// A leader-selected query. - LeaderSelected, + LeaderSelected(LeaderSelectedQueryConfig), } diff --git a/integration_tests/tests/integration/simulation/reproduction.rs b/integration_tests/tests/integration/simulation/reproduction.rs index d27e9203c..52fcb79d8 100644 --- a/integration_tests/tests/integration/simulation/reproduction.rs +++ b/integration_tests/tests/integration/simulation/reproduction.rs @@ -21,8 +21,14 @@ fn successful_collection_time_interval() { install_test_trace_subscriber(); initialize_rustls(); - let collection_job_id = random(); let time_precision = TimePrecision::from_seconds(3600); + let query = Query::TimeInterval( + Interval::new( + Time::from_seconds_since_epoch(1_699_999_200, &time_precision), + Duration::from_seconds(3600, &time_precision), + ) + .unwrap(), + ); let input = Input { is_leader_selected: false, config: Config { @@ -58,17 +64,12 @@ fn successful_collection_time_interval() { Op::LeaderAggregationJobDriver, Op::LeaderGarbageCollector, Op::CollectorStart { - collection_job_id, - query: Query::TimeInterval( - Interval::new( - Time::from_seconds_since_epoch(1_699_999_200, &time_precision), - Duration::from_seconds(3600, &time_precision), - ) - .unwrap(), - ), + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { + query: query.clone(), + }, Op::Upload { report_time: START_TIME.to_time(&time_precision), count: 4, @@ -76,17 +77,10 @@ fn successful_collection_time_interval() { Op::AggregationJobCreator, Op::LeaderAggregationJobDriver, Op::CollectorStart { - collection_job_id, - query: Query::TimeInterval( - Interval::new( - Time::from_seconds_since_epoch(1_699_999_200, &time_precision), - Duration::from_seconds(3600, &time_precision), - ) - .unwrap(), - ), + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { query }, ]), }; let (test_result, metrics) = Simulation::run_with_metrics(input); @@ -129,8 +123,8 @@ fn successful_collection_leader_selected() { install_test_trace_subscriber(); initialize_rustls(); - let collection_job_id = random(); let time_precision = TimePrecision::from_seconds(3600); + let query = Query::LeaderSelected(random()); let input = Input { is_leader_selected: true, config: Config { @@ -166,11 +160,12 @@ fn successful_collection_leader_selected() { Op::LeaderAggregationJobDriver, Op::LeaderGarbageCollector, Op::CollectorStart { - collection_job_id, - query: Query::LeaderSelected, + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { + query: query.clone(), + }, Op::Upload { report_time: START_TIME.to_time(&time_precision), count: 4, @@ -178,11 +173,10 @@ fn successful_collection_leader_selected() { Op::AggregationJobCreator, Op::LeaderAggregationJobDriver, Op::CollectorStart { - collection_job_id, - query: Query::LeaderSelected, + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { query }, ]), }; let (test_result, metrics) = Simulation::run_with_metrics(input); @@ -225,8 +219,14 @@ fn successful_collection_asynchronous() { install_test_trace_subscriber(); initialize_rustls(); - let collection_job_id = random(); let time_precision = TimePrecision::from_seconds(3600); + let query = Query::TimeInterval( + Interval::new( + Time::from_seconds_since_epoch(1_699_999_200, &time_precision), + Duration::from_seconds(3600, &time_precision), + ) + .unwrap(), + ); let input = Input { is_leader_selected: false, config: Config { @@ -277,17 +277,12 @@ fn successful_collection_asynchronous() { Op::LeaderAggregationJobDriver, Op::LeaderGarbageCollector, Op::CollectorStart { - collection_job_id, - query: Query::TimeInterval( - Interval::new( - Time::from_seconds_since_epoch(1_699_999_200, &time_precision), - Duration::from_seconds(3600, &time_precision), - ) - .unwrap(), - ), + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { + query: query.clone(), + }, Op::Upload { report_time: START_TIME.to_time(&time_precision), count: 4, @@ -300,17 +295,10 @@ fn successful_collection_asynchronous() { }, Op::LeaderAggregationJobDriver, Op::CollectorStart { - collection_job_id, - query: Query::TimeInterval( - Interval::new( - Time::from_seconds_since_epoch(1_699_999_200, &time_precision), - Duration::from_seconds(3600, &time_precision), - ) - .unwrap(), - ), + query: query.clone(), }, Op::CollectionJobDriver, - Op::CollectorPoll { collection_job_id }, + Op::CollectorPoll { query }, ]), }; let (test_result, metrics) = Simulation::run_with_metrics(input); @@ -489,7 +477,6 @@ fn repro_abandoned_aggregation_job_batch_mismatch() { install_test_trace_subscriber(); initialize_rustls(); - let collection_job_id = random(); let time_precision = TimePrecision::from_seconds(1000); let input = Input { is_leader_selected: false, @@ -525,7 +512,6 @@ fn repro_abandoned_aggregation_job_batch_mismatch() { }, Op::LeaderAggregationJobDriver, Op::CollectorStart { - collection_job_id, query: Query::TimeInterval( Interval::new( START_TIME.to_time(&time_precision), @@ -571,7 +557,6 @@ fn repro_helper_accumulate_on_retried_request() { }, Op::LeaderAggregationJobDriver, Op::CollectorStart { - collection_job_id: random(), query: Query::TimeInterval( Interval::new( START_TIME.to_time(&time_precision), diff --git a/integration_tests/tests/integration/simulation/run.rs b/integration_tests/tests/integration/simulation/run.rs index 815ae259a..273de38b6 100644 --- a/integration_tests/tests/integration/simulation/run.rs +++ b/integration_tests/tests/integration/simulation/run.rs @@ -150,16 +150,11 @@ impl Simulation { .execute_collection_job_driver_response_error() .await } - Op::CollectorStart { - collection_job_id, - query, - } => { - simulation - .execute_collector_start(collection_job_id, query) - .await + Op::CollectorStart { query } => { + simulation.execute_collector_start(query).await } - Op::CollectorPoll { collection_job_id } => { - simulation.execute_collector_poll(collection_job_id).await + Op::CollectorPoll { query } => { + simulation.execute_collector_poll(query).await } }; info!(elapsed = ?start.elapsed(), "finished operation"); @@ -430,44 +425,39 @@ impl Simulation { result } - async fn execute_collector_start( - &mut self, - collection_job_id: &CollectionJobId, - query: &Query, - ) -> ControlFlow { + async fn execute_collector_start(&mut self, query: &Query) -> ControlFlow { match query { Query::TimeInterval(interval) => { - let query = janus_messages::Query::new_time_interval(*interval); match self .components .collector - .collection(query, &()) + .collection(janus_messages::Query::new_time_interval(*interval), &()) .start() .await { Ok(collection_job) => { self.state .collection_jobs_time_interval - .insert(*collection_job_id, collection_job); + .insert(query.clone(), collection_job); } Err(error) => info!(?error, "collector error"), } } - Query::LeaderSelected => { - let query = janus_messages::Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ); + Query::LeaderSelected(idempotency_key) => { match self .components .collector - .collection(query, &()) + .collection( + janus_messages::Query::new_leader_selected(idempotency_key.clone()), + &(), + ) .start() .await { Ok(collection_job) => { self.state .collection_jobs_leader_selected - .insert(*collection_job_id, collection_job); + .insert(query.clone(), collection_job); } Err(error) => info!(?error, "collector error"), } @@ -476,15 +466,8 @@ impl Simulation { ControlFlow::Continue(()) } - async fn execute_collector_poll( - &mut self, - collection_job_id: &CollectionJobId, - ) -> ControlFlow { - if let Some(collection_job) = self - .state - .collection_jobs_time_interval - .get(collection_job_id) - { + async fn execute_collector_poll(&mut self, query: &Query) -> ControlFlow { + if let Some(collection_job) = self.state.collection_jobs_time_interval.get(query) { let result = self.components.collector.poll_once(collection_job).await; match result { Ok(PollResult::CollectionResult(collection)) => { @@ -494,7 +477,7 @@ impl Simulation { let old_opt = self .state .aggregate_results_time_interval - .insert(*collection_job_id, collection); + .insert(*collection_job.collection_job_id(), collection); if let Some(old_collection) = old_opt { if report_count != old_collection.report_count() || &interval != old_collection.interval() @@ -508,11 +491,7 @@ impl Simulation { Ok(PollResult::NotReady(_)) => {} Err(error) => info!(?error, "collector error"), } - } else if let Some(collection_job) = self - .state - .collection_jobs_leader_selected - .get(collection_job_id) - { + } else if let Some(collection_job) = self.state.collection_jobs_leader_selected.get(query) { let result = self.components.collector.poll_once(collection_job).await; match result { Ok(PollResult::CollectionResult(collection)) => { @@ -522,7 +501,7 @@ impl Simulation { let old_opt = self .state .aggregate_results_leader_selected - .insert(*collection_job_id, collection); + .insert(*collection_job.collection_job_id(), collection); if let Some(old_collection) = old_opt { if report_count != old_collection.report_count() || &interval != old_collection.interval() @@ -566,10 +545,8 @@ pub(super) struct State { pub(super) runtime_manager: TestRuntimeManager<&'static str>, pub(super) vdaf_instance: VdafInstance, pub(super) vdaf: Prio3Histogram, - pub(super) collection_jobs_time_interval: - HashMap>, - pub(super) collection_jobs_leader_selected: - HashMap>, + pub(super) collection_jobs_time_interval: HashMap>, + pub(super) collection_jobs_leader_selected: HashMap>, pub(super) aggregate_results_time_interval: HashMap>>, pub(super) aggregate_results_leader_selected: HashMap>>, pub(super) next_measurement: usize, diff --git a/interop_binaries/src/commands/janus_interop_collector.rs b/interop_binaries/src/commands/janus_interop_collector.rs index 72f442a19..bb8b722ab 100644 --- a/interop_binaries/src/commands/janus_interop_collector.rs +++ b/interop_binaries/src/commands/janus_interop_collector.rs @@ -437,9 +437,7 @@ async fn handle_collection_start( handle_collect_generic( http_client, task_state, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(random()), vdaf, &agg_param, |result| AggregationResult::Number(NumberAsString((*result).into())), @@ -453,9 +451,7 @@ async fn handle_collection_start( handle_collect_generic( http_client, task_state, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(random()), vdaf, &agg_param, |result| AggregationResult::Number(NumberAsString(u128::from(*result))), @@ -477,9 +473,7 @@ async fn handle_collection_start( handle_collect_generic( http_client, task_state, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(random()), vdaf, &agg_param, |result| { @@ -507,9 +501,7 @@ async fn handle_collection_start( handle_collect_generic( http_client, task_state, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(random()), vdaf, &agg_param, |result: &Vec| { @@ -538,9 +530,7 @@ async fn handle_collection_start( handle_collect_generic( http_client, task_state, - Query::new_leader_selected( - janus_messages::batch_mode::LeaderSelectedQueryConfig::test_default(), - ), + Query::new_leader_selected(random()), vdaf, &agg_param, |result| {