add hybrid_protocol function, unimplemented (private-attribution#1375)

* add hybrid_protocol function, unimplemented

* remove traitbounds for now, add them back as needed

* update hybrid protocol comment

* add comment about copy pasted BreakdownKey trait

ipa-core/src/protocol/hybrid/mod.rs

@@ -1 +1,79 @@
 pub(crate) mod step;
+
+use crate::{
+    error::Error,
+    ff::{
+        boolean_array::{BooleanArray, BA5, BA8},
+        U128Conversions,
+    },
+    helpers::query::DpMechanism,
+    protocol::{
+        context::{ShardedContext, UpgradableContext},
+        ipa_prf::{oprf_padding::PaddingParameters, shuffle::Shuffle},
+    },
+    report::hybrid::IndistinguishableHybridReport,
+    secret_sharing::replicated::semi_honest::AdditiveShare as Replicated,
+};
+
+// In theory, we could support (runtime-configured breakdown count) ≤ (compile-time breakdown count)
+// ≤ 2^|bk|, with all three values distinct, but at present, there is no runtime configuration and
+// the latter two must be equal. The implementation of `move_single_value_to_bucket` does support a
+// runtime-specified count via the `breakdown_count` parameter, and implements a runtime check of
+// its value.
+//
+// It would usually be more appropriate to make `MAX_BREAKDOWNS` an associated constant rather than
+// a const parameter. However, we want to use it to enforce a correct pairing of the `BK` type
+// parameter and the `B` const parameter, and specifying a constraint like
+// `BreakdownKey<MAX_BREAKDOWNS = B>` on an associated constant is not currently supported. (Nor is
+// supplying an associated constant `<BK as BreakdownKey>::MAX_BREAKDOWNS` as the value of a const
+// parameter.) Structured the way we have it, it probably doesn't make sense to use the
+// `BreakdownKey` trait in places where the `B` const parameter is not already available.
+//
+// These could be imported from src/protocl/ipa_prf/mod.rs
+// however we've copy/pasted them here with the intention of deleting that file [TODO]
+pub trait BreakdownKey<const MAX_BREAKDOWNS: usize>: BooleanArray + U128Conversions {}
+impl BreakdownKey<32> for BA5 {}
+impl BreakdownKey<256> for BA8 {}
+
+/// The Hybrid Protocol
+///
+/// This protocol takes in a [`Vec<IndistinguishableHybridReport<BK, V>>`]
+/// and aggregates it into a summary report. `HybridReport`s are either
+/// impressions or conversion. The protocol joins these based on their matchkeys,
+/// sums the values from conversions grouped by the breakdown key on impressions.
+/// To accomplish this, hte protocol performs the follwoing steps
+/// 1. Generates a random number of "dummy records" (needed to mask the information that will
+///    be revealed in step 4, and thereby provide a differential privacy guarantee on
+///    that information leakage)
+/// 2. Shuffles the input
+/// 3. Computes an OPRF of these elliptic curve points and reveals this "pseudonym"
+/// 4. Groups together rows with the same OPRF and sums both the breakdown keys and values.
+/// 5. Generates a random number of "dummy records" (needed to mask the information that will
+///    be revealed in step 7)
+/// 6. Shuffles the input
+/// 7. Reveals breakdown keys
+/// 8. Sums the values by breakdown keys
+/// 9. Adds random noise to the total value for each breakdown key (to provide a
+///    differential privacy guarantee)
+///
+/// # Errors
+/// Propagates errors from config issues or while running the protocol
+/// # Panics
+/// Propagates errors from config issues or while running the protocol
+pub async fn hybrid_protocol<'ctx, C, BK, V, HV, const SS_BITS: usize, const B: usize>(
+    _ctx: C,
+    input_rows: Vec<IndistinguishableHybridReport<BK, V>>,
+    _dp_params: DpMechanism,
+    _dp_padding_params: PaddingParameters,
+) -> Result<Vec<Replicated<HV>>, Error>
+where
+    C: UpgradableContext + 'ctx + Shuffle + ShardedContext,
+    BK: BreakdownKey<B>,
+    V: BooleanArray + U128Conversions,
+    HV: BooleanArray + U128Conversions,
+{
+    if input_rows.is_empty() {
+        return Ok(vec![Replicated::ZERO; B]);
+    }
+    unimplemented!("protocol::hybrid::hybrid_protocol is not fully implemented")
+}

ipa-core/src/query/runner/hybrid.rs

@@ -4,18 +4,26 @@ use futures::{stream::iter, StreamExt, TryStreamExt};
 
 use crate::{
     error::Error,
-    ff::boolean_array::{BA20, BA3, BA8},
+    ff::{
+        boolean_array::{BooleanArray, BA20, BA3, BA8},
+        U128Conversions,
+    },
     helpers::{
-        query::{HybridQueryParams, QuerySize},
+        query::{DpMechanism, HybridQueryParams, QuerySize},
         BodyStream, LengthDelimitedStream,
     },
     hpke::PrivateKeyRegistry,
-    protocol::{context::ShardedContext, hybrid::step::HybridStep, step::ProtocolStep::Hybrid},
+    protocol::{
+        context::{ShardedContext, UpgradableContext},
+        hybrid::{hybrid_protocol, step::HybridStep},
+        ipa_prf::{oprf_padding::PaddingParameters, shuffle::Shuffle},
+        step::ProtocolStep::Hybrid,
+    },
     query::runner::reshard_tag::reshard_aad,
     report::hybrid::{
         EncryptedHybridReport, IndistinguishableHybridReport, UniqueTag, UniqueTagValidator,
     },
-    secret_sharing::{replicated::semi_honest::AdditiveShare as ReplicatedShare, SharedValue},
+    secret_sharing::replicated::semi_honest::AdditiveShare as Replicated,
 };
 
 #[allow(dead_code)]
@@ -25,25 +33,30 @@ pub struct Query<C, HV, R: PrivateKeyRegistry> {
     phantom_data: PhantomData<(C, HV)>,
 }
 
-impl<C, HV: SharedValue, R: PrivateKeyRegistry> Query<C, HV, R>
-where
-    C: ShardedContext,
-{
+#[allow(dead_code)]
+impl<C, HV, R: PrivateKeyRegistry> Query<C, HV, R> {
     pub fn new(query_params: HybridQueryParams, key_registry: Arc<R>) -> Self {
         Self {
             config: query_params,
             key_registry,
             phantom_data: PhantomData,
         }
     }
+}
 
+impl<C, HV, R> Query<C, HV, R>
+where
+    C: UpgradableContext + Shuffle + ShardedContext,
+    HV: BooleanArray + U128Conversions,
+    R: PrivateKeyRegistry,
+{
     #[tracing::instrument("hybrid_query", skip_all, fields(sz=%query_size))]
     pub async fn execute(
         self,
         ctx: C,
         query_size: QuerySize,
         input_stream: BodyStream,
-    ) -> Result<Vec<ReplicatedShare<HV>>, Error> {
+    ) -> Result<Vec<Replicated<HV>>, Error> {
         let Self {
             config,
             key_registry,
@@ -89,10 +102,34 @@ where
             .check_duplicates(&resharded_tags)
             .unwrap();
 
-        let _indistinguishable_reports: Vec<IndistinguishableHybridReport<BA8, BA3>> =
+        let indistinguishable_reports: Vec<IndistinguishableHybridReport<BA8, BA3>> =
             decrypted_reports.into_iter().map(Into::into).collect();
 
-        unimplemented!("query::runnner::HybridQuery.execute is not fully implemented")
+        let dp_params: DpMechanism = match config.with_dp {
+            0 => DpMechanism::NoDp,
+            _ => DpMechanism::DiscreteLaplace {
+                epsilon: config.epsilon,
+            },
+        };
+
+        #[cfg(feature = "relaxed-dp")]
+        let padding_params = PaddingParameters::relaxed();
+        #[cfg(not(feature = "relaxed-dp"))]
+        let padding_params = PaddingParameters::default();
+
+        match config.per_user_credit_cap {
+            1 => hybrid_protocol::<_, BA8, BA3, HV, 1, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            2 | 4 => hybrid_protocol::<_, BA8, BA3, HV, 2, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            8 => hybrid_protocol::<_, BA8, BA3, HV, 3, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            16 => hybrid_protocol::<_, BA8, BA3, HV, 4, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            32 => hybrid_protocol::<_, BA8, BA3, HV, 5, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            64 => hybrid_protocol::<_, BA8, BA3, HV, 6, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            128 => hybrid_protocol::<_, BA8, BA3, HV, 7, 256>(ctx, indistinguishable_reports, dp_params, padding_params).await,
+            _ => panic!(
+                "Invalid value specified for per-user cap: {:?}. Must be one of 1, 2, 4, 8, 16, 32, 64, or 128.",
+                config.per_user_credit_cap
+            ),
+        }
     }
 }
 
@@ -219,7 +256,7 @@ mod tests {
     // placeholder until the protocol is complete. can be updated to make sure we
     // get to the unimplemented() call
     #[should_panic(
-        expected = "not implemented: query::runnner::HybridQuery.execute is not fully implemented"
+        expected = "not implemented: protocol::hybrid::hybrid_protocol is not fully implemented"
     )]
     async fn encrypted_hybrid_reports() {
         // While this test currently checks for an unimplemented panic it is