DAP-05 ping-pong topology

This change implements the DAP-05 ping-pong topology in which
aggregators take turns preprocessing prepare shares into prepare
messages. While this topology first appeared in DAP-05, this
implementation follows the changes in [1], which should appear in
DAP-06.

This change depends on the implementation of the VDAF ping-pong topology
added to crate `prio` in [2], which in turn conforms to the
specification added after VDAF-06 and further tweaked in [3] (we expect
this to be published soon as VDAF-07).

This commit makes some changes to what intermediate values are stored by
aggregators. In the case where an aggregator is continuing, it will have
computed a prepare state, a prepare message for the current round and a
prepare share for the next round. The existing implementation would
store all three objects in the database, significantly increasing the
per-report storage requirements. In particular, this makes things worse
for the Helper, which previously never needed to store a prepare share
because the Leader always took responsibility for combining prepare
shares.

To mitigate this, we instead have aggregators store a
`prio::ping_pong::topology::Transition`, which will contain a prepare
state and a prepare message (both of which are generally much smaller
than prepare shares), from which the next prepare state and importantly
prepare share can be recomputed.

The main benefit of this change is to reduce how many round trips
between aggregators are needed to prepare reports. Quite a few tests
used Prio3 but depended on having the leader or helper in the `Waiting`
state after running aggregation initialization. Accordingly, those tests
are changed to run Poplar1, which now takes 2 rounds.

[1]: ietf-wg-ppm/draft-ietf-ppm-dap#494
[2]: divviup/libprio-rs#683
[3]: cfrg/draft-irtf-cfrg-vdaf#281

Part of #1669

Cargo.toml

@@ -42,7 +42,9 @@ janus_messages = { version = "0.5", path = "messages" }
 k8s-openapi = { version = "0.18.0", features = ["v1_24"] }  # keep this version in sync with what is referenced by the indirect dependency via `kube`
 kube = { version = "0.82.2", default-features = false, features = ["client", "rustls-tls"] }
 opentelemetry = { version = "0.20", features = ["metrics"] }
-prio = { version = "0.14.1", features = ["multithreaded"] }
+# TODO(timg): go back to a released version of prio
+#prio = { version = "0.14.1", features = ["multithreaded"] }
+prio = { git = "https://github.com/divviup/libprio-rs", branch = "timg/ping-pong-topology", features = ["multithreaded", "experimental"] }
 serde = { version = "1.0.185", features = ["derive"] }
 serde_json = "1.0.105"
 serde_test = "1.0.175"

aggregator/src/aggregator/accumulator.rs

@@ -26,7 +26,7 @@ use std::{
 /// Accumulates output shares in memory and eventually flushes accumulations to a datastore. We
 /// accumulate output shares into a [`HashMap`] mapping the batch identifier at which the batch
 /// interval begins to the accumulated aggregate share, report count and checksum.
-#[derive(Derivative)]
+#[derive(Clone, Derivative)]
 #[derivative(Debug)]
 pub struct Accumulator<
     const SEED_SIZE: usize,
@@ -40,7 +40,7 @@ pub struct Accumulator<
     aggregations: HashMap<Q::BatchIdentifier, BatchData<SEED_SIZE, Q, A>>,
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 struct BatchData<
     const SEED_SIZE: usize,
     Q: AccumulableQueryType,