Repository: cloudflare/daphne
Pull request #353 (2023-06-20 16:40:27)

Improve kv access patterns by not listing all keys
## Commits
- move HpkeConfig into hpke module
    I was finding it very confusing that some parts of hpke were not in the hpke module, so I reorganized this.
- Use fmt::Display instead of ToString
    It is more ideomatic and slightly more efficient to use Display over ToString
- Change hpke config schema in kv
    See bellow for longer explanation.
- Configure the hpke receiver config from the tests
    See commit message for more details.

## New Kv Schema
Instead of having one key per config like so
- hpke_receiver_config/version/v02/config_id/24 => `{ public_key, private_key }`
- hpke_receiver_config/version/v02/config_id/25 => `{ public_key, private_key }`
- hpke_receiver_config/version/v02/config_id/26 => `{ public_key, private_key }`
- hpke_receiver_config/version/v04/config_id/45 => `{ public_key, private_key }`

which requires listing the keys from kv to find the available configs, this commit changes the schema to store the configs in a single key per version in the following schema:
- hpke_receiver_config/v02 =>
    ```
    {
        24: { public, private },
        25; { public, private },
        26; { public, private },
    }
    ```
- hpke_receiver_config/v04 =>
    ```
    {
        45: { public, private },
    }
    ```

This avoids the `kv::list` operation, and makes the implementation of key rotation in the future feasible.

## Removed automatic generation
I also decided to remove the automatic generation of configurations since it's a code path that was never used in production and key configurations should be managed from outside the worker. On top of that it relied on the `set_if_not_exists` method whose implementation has a TOCTOU bug


Patch:
diff --git a/daphne/dapf/src/bin/dapf.rs b/daphne/dapf/src/bin/dapf.rs
index 8f18666..775367a 100644
--- a/daphne/dapf/src/bin/dapf.rs
+++ b/daphne/dapf/src/bin/dapf.rs
@@ -6,8 +6,8 @@ use clap::{builder::PossibleValue, Parser, Subcommand, ValueEnum};
 use daphne::{
     constants::DapMediaType,
     error::aborts::ProblemDetails,
-    hpke::HpkeReceiverConfig,
-    messages::{BatchSelector, Collection, CollectionReq, HpkeConfig, HpkeKemId, Query, TaskId},
+    hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
+    messages::{BatchSelector, Collection, CollectionReq, Query, TaskId},
     DapMeasurement, DapVersion, VdafConfig,
 };
 use prio::codec::{Decode, ParameterizedDecode, ParameterizedEncode};
diff --git a/daphne/src/hpke.rs b/daphne/src/hpke.rs
index 3d13ea3..26f48c6 100644
--- a/daphne/src/hpke.rs
+++ b/daphne/src/hpke.rs
@@ -13,10 +13,7 @@ use hpke_rs_rust_crypto::HpkeRustCrypto as ImplHpkeCrypto;
 
 use crate::{
     fatal_error,
-    messages::{
-        decode_u16_bytes, encode_u16_bytes, HpkeAeadId, HpkeCiphertext, HpkeConfig, HpkeKdfId,
-        HpkeKemId, TaskId, TransitionFailure,
-    },
+    messages::{decode_u16_bytes, encode_u16_bytes, HpkeCiphertext, TaskId, TransitionFailure},
     DapError, DapVersion,
 };
 use async_trait::async_trait;
@@ -24,6 +21,12 @@ use prio::codec::{CodecError, Decode, Encode};
 use serde::{Deserialize, Serialize};
 use std::io::Cursor;
 
+// Various algorithm constants
+const KEM_ID_X25519_HKDF_SHA256: u16 = 0x0020;
+const KEM_ID_P256_HKDF_SHA256: u16 = 0x0010;
+const KDF_ID_HKDF_SHA256: u16 = 0x0001;
+const AEAD_ID_AES128GCM: u16 = 0x0001;
+
 impl From<HpkeError> for DapError {
     fn from(_e: HpkeError) -> Self {
         Self::Transition(TransitionFailure::HpkeDecryptError)
@@ -60,6 +63,104 @@ fn check_suite<T: HpkeCrypto>(
     }
 }
 
+/// Codepoint for KEM schemes compatible with HPKE.
+#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum HpkeKemId {
+    P256HkdfSha256,
+    X25519HkdfSha256,
+    NotImplemented(u16),
+}
+
+impl From<HpkeKemId> for u16 {
+    fn from(kem_id: HpkeKemId) -> Self {
+        match kem_id {
+            HpkeKemId::P256HkdfSha256 => KEM_ID_P256_HKDF_SHA256,
+            HpkeKemId::X25519HkdfSha256 => KEM_ID_X25519_HKDF_SHA256,
+            HpkeKemId::NotImplemented(x) => x,
+        }
+    }
+}
+
+impl From<u16> for HpkeKemId {
+    fn from(value: u16) -> Self {
+        match value {
+            KEM_ID_P256_HKDF_SHA256 => Self::P256HkdfSha256,
+            KEM_ID_X25519_HKDF_SHA256 => Self::X25519HkdfSha256,
+            x => Self::NotImplemented(x),
+        }
+    }
+}
+
+/// Codepoint for KDF schemes compatible with HPKE.
+#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum HpkeKdfId {
+    HkdfSha256,
+    NotImplemented(u16),
+}
+
+impl From<HpkeKdfId> for u16 {
+    fn from(kdf_id: HpkeKdfId) -> Self {
+        match kdf_id {
+            HpkeKdfId::HkdfSha256 => KDF_ID_HKDF_SHA256,
+            HpkeKdfId::NotImplemented(x) => x,
+        }
+    }
+}
+
+impl From<u16> for HpkeKdfId {
+    fn from(value: u16) -> Self {
+        match value {
+            KDF_ID_HKDF_SHA256 => Self::HkdfSha256,
+            x => Self::NotImplemented(x),
+        }
+    }
+}
+
+/// Codepoint for AEAD schemes compatible with HPKE.
+#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum HpkeAeadId {
+    Aes128Gcm,
+    NotImplemented(u16),
+}
+
+impl From<HpkeAeadId> for u16 {
+    fn from(aead_id: HpkeAeadId) -> Self {
+        match aead_id {
+            HpkeAeadId::Aes128Gcm => AEAD_ID_AES128GCM,
+            HpkeAeadId::NotImplemented(x) => x,
+        }
+    }
+}
+
+impl From<u16> for HpkeAeadId {
+    fn from(value: u16) -> Self {
+        match value {
+            AEAD_ID_AES128GCM => Self::Aes128Gcm,
+            x => Self::NotImplemented(x),
+        }
+    }
+}
+
+/// The HPKE public key configuration of a Server.
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
+pub struct HpkeConfig {
+    pub id: u8,
+    pub kem_id: HpkeKemId,
+    pub kdf_id: HpkeKdfId,
+    pub aead_id: HpkeAeadId,
+    #[serde(with = "HpkePublicKeySerde")]
+    pub public_key: HpkePublicKey,
+}
+
+impl AsRef<HpkeConfig> for HpkeConfig {
+    fn as_ref(&self) -> &Self {
+        self
+    }
+}
+
 impl HpkeConfig {
     /// Encrypt `plaintext` with info string `info` and associated data `aad` using this HPKE
     /// configuration. The return values are the encapsulated key and the ciphertext.
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 5441266..6d569e1 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -42,8 +42,8 @@ use crate::{
     hpke::HpkeReceiverConfig,
     messages::{
         AggregationJobId, BatchId, BatchSelector, Collection, CollectionJobId,
-        Draft02AggregationJobId, Duration, HpkeConfig, HpkeKemId, Interval, PartialBatchSelector,
-        ReportId, ReportMetadata, TaskId, Time,
+        Draft02AggregationJobId, Duration, Interval, PartialBatchSelector, ReportId,
+        ReportMetadata, TaskId, Time,
     },
     taskprov::TaskprovVersion,
     vdaf::{
@@ -54,6 +54,7 @@ use crate::{
 };
 use constants::DapMediaType;
 pub use error::DapError;
+use hpke::{HpkeConfig, HpkeKemId};
 use prio::{
     codec::{Decode, Encode},
     vdaf::Aggregatable as AggregatableTrait,
@@ -160,7 +161,7 @@ impl DapGlobalConfig {
         &self,
         first_config_id: u8,
     ) -> impl Iterator<Item = Result<HpkeReceiverConfig, DapError>> {
-        assert!(self.supported_hpke_kems.len() <= 256);
+        assert!(self.supported_hpke_kems.len() <= u8::MAX.into());
         let kem_ids = self.supported_hpke_kems.clone();
         kem_ids.into_iter().enumerate().map(move |(i, kem_id)| {
             let (config_id, _overflowed) = first_config_id.overflowing_add(i as u8);
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index 0c22533..233a029 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -3,7 +3,11 @@
 
 //! Messages in the DAP protocol.
 
-use crate::{fatal_error, hpke::HpkePublicKeySerde, DapError, DapVersion};
+use crate::{
+    fatal_error,
+    hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId},
+    DapError, DapVersion,
+};
 use base64::engine::{general_purpose::URL_SAFE_NO_PAD, Engine};
 use hpke_rs::HpkePublicKey;
 use prio::codec::{
@@ -18,12 +22,6 @@ use std::{
     io::{Cursor, Read},
 };
 
-// Various algorithm constants
-const KEM_ID_X25519_HKDF_SHA256: u16 = 0x0020;
-const KEM_ID_P256_HKDF_SHA256: u16 = 0x0010;
-const KDF_ID_HKDF_SHA256: u16 = 0x0001;
-const AEAD_ID_AES128GCM: u16 = 0x0001;
-
 // Query types
 const QUERY_TYPE_TIME_INTERVAL: u8 = 0x01;
 const QUERY_TYPE_FIXED_SIZE: u8 = 0x02;
@@ -956,23 +954,10 @@ impl Decode for AggregateShare {
     }
 }
 
-/// Codepoint for KEM schemes compatible with HPKE.
-#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
-#[serde(rename_all = "snake_case")]
-pub enum HpkeKemId {
-    P256HkdfSha256,
-    X25519HkdfSha256,
-    NotImplemented(u16),
-}
-
-impl From<HpkeKemId> for u16 {
-    fn from(kem_id: HpkeKemId) -> Self {
-        match kem_id {
-            HpkeKemId::P256HkdfSha256 => KEM_ID_P256_HKDF_SHA256,
-            HpkeKemId::X25519HkdfSha256 => KEM_ID_X25519_HKDF_SHA256,
-            HpkeKemId::NotImplemented(x) => x,
-        }
-    }
+/// A list of HPKE public key configurations.
+#[derive(Clone, Debug, PartialEq)]
+pub struct HpkeConfigList {
+    pub hpke_configs: Vec<HpkeConfig>,
 }
 
 impl Encode for HpkeKemId {
@@ -983,28 +968,7 @@ impl Encode for HpkeKemId {
 
 impl Decode for HpkeKemId {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        match u16::decode(bytes)? {
-            x if x == KEM_ID_P256_HKDF_SHA256 => Ok(Self::P256HkdfSha256),
-            x if x == KEM_ID_X25519_HKDF_SHA256 => Ok(Self::X25519HkdfSha256),
-            x => Ok(Self::NotImplemented(x)),
-        }
-    }
-}
-
-/// Codepoint for KDF schemes compatible with HPKE.
-#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
-#[serde(rename_all = "snake_case")]
-pub enum HpkeKdfId {
-    HkdfSha256,
-    NotImplemented(u16),
-}
-
-impl From<HpkeKdfId> for u16 {
-    fn from(kdf_id: HpkeKdfId) -> Self {
-        match kdf_id {
-            HpkeKdfId::HkdfSha256 => KDF_ID_HKDF_SHA256,
-            HpkeKdfId::NotImplemented(x) => x,
-        }
+        Ok(u16::decode(bytes)?.into())
     }
 }
 
@@ -1016,27 +980,7 @@ impl Encode for HpkeKdfId {
 
 impl Decode for HpkeKdfId {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        match u16::decode(bytes)? {
-            x if x == KDF_ID_HKDF_SHA256 => Ok(Self::HkdfSha256),
-            x => Ok(Self::NotImplemented(x)),
-        }
-    }
-}
-
-/// Codepoint for AEAD schemes compatible with HPKE.
-#[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
-#[serde(rename_all = "snake_case")]
-pub enum HpkeAeadId {
-    Aes128Gcm,
-    NotImplemented(u16),
-}
-
-impl From<HpkeAeadId> for u16 {
-    fn from(aead_id: HpkeAeadId) -> Self {
-        match aead_id {
-            HpkeAeadId::Aes128Gcm => AEAD_ID_AES128GCM,
-            HpkeAeadId::NotImplemented(x) => x,
-        }
+        Ok(u16::decode(bytes)?.into())
     }
 }
 
@@ -1048,27 +992,7 @@ impl Encode for HpkeAeadId {
 
 impl Decode for HpkeAeadId {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        match u16::decode(bytes)? {
-            x if x == AEAD_ID_AES128GCM => Ok(Self::Aes128Gcm),
-            x => Ok(Self::NotImplemented(x)),
-        }
-    }
-}
-
-/// The HPKE public key configuration of a Server.
-#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
-pub struct HpkeConfig {
-    pub id: u8,
-    pub kem_id: HpkeKemId,
-    pub kdf_id: HpkeKdfId,
-    pub aead_id: HpkeAeadId,
-    #[serde(with = "HpkePublicKeySerde")]
-    pub public_key: HpkePublicKey,
-}
-
-impl AsRef<HpkeConfig> for HpkeConfig {
-    fn as_ref(&self) -> &Self {
-        self
+        Ok(u16::decode(bytes)?.into())
     }
 }
 
@@ -1094,12 +1018,6 @@ impl Decode for HpkeConfig {
     }
 }
 
-/// A list of HPKE public key configurations.
-#[derive(Clone, Debug, PartialEq)]
-pub struct HpkeConfigList {
-    pub hpke_configs: Vec<HpkeConfig>,
-}
-
 impl Encode for HpkeConfigList {
     fn encode(&self, bytes: &mut Vec<u8>) {
         encode_u16_items(bytes, &(), &self.hpke_configs);
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 7185533..28c7548 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -7,12 +7,12 @@ use crate::{
     audit_log::{AggregationJobAuditAction, AuditLog},
     constants::DapMediaType,
     fatal_error,
-    hpke::HpkeDecrypter,
+    hpke::{HpkeConfig, HpkeDecrypter},
     messages::{
         constant_time_eq, decode_base64url, AggregateShare, AggregateShareReq,
         AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp, BatchId,
-        BatchSelector, Collection, CollectionJobId, CollectionReq, HpkeConfig, HpkeConfigList,
-        Interval, PartialBatchSelector, Query, Report, ReportId, ReportMetadata, TaskId, Time,
+        BatchSelector, Collection, CollectionJobId, CollectionReq, HpkeConfigList, Interval,
+        PartialBatchSelector, Query, Report, ReportId, ReportMetadata, TaskId, Time,
         TransitionFailure, TransitionVar,
     },
     metrics::{DaphneMetrics, DaphneRequestType},
diff --git a/daphne/src/taskprov.rs b/daphne/src/taskprov.rs
index b69e80e..9fd325f 100644
--- a/daphne/src/taskprov.rs
+++ b/daphne/src/taskprov.rs
@@ -3,10 +3,11 @@
 
 use crate::{
     fatal_error,
+    hpke::HpkeConfig,
     messages::{
         decode_base64url_vec,
         taskprov::{QueryConfigVar, TaskConfig, VdafType, VdafTypeVar},
-        Extension, HpkeConfig, ReportMetadata, TaskId,
+        Extension, ReportMetadata, TaskId,
     },
     vdaf::VdafVerifyKey,
     DapAbort, DapError, DapQueryConfig, DapRequest, DapTaskConfig, DapVersion, Prio3Config,
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index d97d135..b214b44 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -7,12 +7,12 @@
 use crate::{
     error::DapAbort,
     fatal_error,
-    hpke::HpkeDecrypter,
+    hpke::{HpkeConfig, HpkeDecrypter},
     messages::{
         encode_u32_bytes, AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp,
-        BatchSelector, Extension, HpkeCiphertext, HpkeConfig, PartialBatchSelector,
-        PlaintextInputShare, Report, ReportId, ReportMetadata, ReportShare, TaskId, Time,
-        Transition, TransitionFailure, TransitionVar,
+        BatchSelector, Extension, HpkeCiphertext, PartialBatchSelector, PlaintextInputShare,
+        Report, ReportId, ReportMetadata, ReportShare, TaskId, Time, Transition, TransitionFailure,
+        TransitionVar,
     },
     metrics::ContextualizedDaphneMetrics,
     vdaf::{
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index 8401d3b..43a7a4d 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -24,10 +24,9 @@ use daphne::{
     constants::DapMediaType,
     error::DapAbort,
     fatal_error,
-    hpke::HpkeReceiverConfig,
+    hpke::{HpkeConfig, HpkeReceiverConfig},
     messages::{
-        decode_base64url_vec, AggregationJobId, BatchId, CollectionJobId, HpkeConfig,
-        ReportMetadata, TaskId,
+        decode_base64url_vec, AggregationJobId, BatchId, CollectionJobId, ReportMetadata, TaskId,
     },
     DapError, DapGlobalConfig, DapQueryConfig, DapRequest, DapResource, DapResponse, DapTaskConfig,
     DapVersion, Prio3Config, VdafConfig,
@@ -43,6 +42,7 @@ use serde::{Deserialize, Serialize};
 use std::{
     borrow::Cow,
     collections::HashMap,
+    fmt::Display,
     io::Cursor,
     sync::{Arc, RwLock},
     time::Duration,
@@ -50,7 +50,7 @@ use std::{
 use tracing::{error, info, trace};
 use worker::{kv::KvStore, *};
 
-pub(crate) const KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG: &str = "hpke_receiver_config";
+const KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG_SET: &str = "hpke_receiver_config_set";
 pub(crate) const KV_KEY_PREFIX_BEARER_TOKEN_LEADER: &str = "bearer_token/leader/task";
 pub(crate) const KV_KEY_PREFIX_BEARER_TOKEN_COLLECTOR: &str = "bearer_token/collector/task";
 pub(crate) const KV_KEY_PREFIX_TASK_CONFIG: &str = "config/task";
@@ -360,6 +360,8 @@ impl DaphneWorkerConfig {
     }
 }
 
+pub(crate) type HpkeRecieverConfigSet = HashMap<u8, HpkeReceiverConfig>;
+
 /// Daphne-Worker per-isolate state, which may be used by multiple requests. Includes long-lived configuration,
 /// cached responses from KV, etc.
 pub(crate) struct DaphneWorkerIsolateState {
@@ -370,7 +372,7 @@ pub(crate) struct DaphneWorkerIsolateState {
 
     /// Cached HPKE receiver config. This will be populated when Daphne-Worker obtains an HPKE
     /// receiver config for the first time from Cloudflare KV.
-    hpke_receiver_configs: Arc<RwLock<HashMap<HpkeReceiverKvKey, HpkeReceiverConfig>>>,
+    hpke_receiver_configs: Arc<RwLock<HashMap<DapVersion, HpkeRecieverConfigSet>>>,
 
     /// Laeder bearer token per task.
     pub(crate) leader_bearer_tokens: Arc<RwLock<HashMap<TaskId, BearerToken>>>,
@@ -565,39 +567,16 @@ impl<'srv> DaphneWorker<'srv> {
         Ok(None)
     }
 
-    /// In memory cache on-top of KV to avoid hitting KV API limits.
-    // NOTE: We shouldn't return guards from this function as they could be held across await points.
-    //
-    // Locks in wasm are assumed to be single threaded without concurrency:
-    // https://github.com/rust-lang/rust/blob/6f8c0557e0b73c73a8a7163a15f4a5a3feca7d5c/library/std/src/sys/unsupported/locks/rwlock.rs#L4
-    //
-    // However, with Cloudflare Workers the situation is not so straightforward.
-    //
-    // A single isolate can process multiple requests concurrently (not in parallel), so holding
-    // ReadGuards across await points is not OK.
-    //
-    // There are some official docs that refer to the behaviour:
-    // https://developers.cloudflare.com/workers/learning/how-workers-works/#distributed-execution
-    //
-    // The errors seen are:
-    //
-    //   Error: The script will never generate a response.
-    //
-    //   and
-    //
-    //   Memory access out of bounds
-    //
-    // They only seem to happen under load, likely because the Workers runtime will readily spawn
-    // additional isolates.
-    async fn kv_get_cached<'req, K, V>(
+    async fn kv_get_cached_mapped<'req, K, V, R>(
         &self,
         map: &'srv Arc<RwLock<HashMap<K, V>>>,
         kv_key_prefix: &str,
         kv_key_suffix: Cow<'req, K>,
-    ) -> Result<Option<KvPair<'req, K, V>>>
+        mapper: impl FnOnce(KvPair<'req, K, &V>) -> R,
+    ) -> Result<Option<R>>
     where
-        K: Clone + Eq + std::hash::Hash + ToString,
-        V: Clone + for<'de> Deserialize<'de>,
+        K: Clone + Eq + std::hash::Hash + Display,
+        V: for<'de> Deserialize<'de>,
     {
         // If the value is cached, then return immediately.
         {
@@ -606,15 +585,18 @@ impl<'srv> DaphneWorker<'srv> {
                 .map_err(|e| Error::RustError(format!("Failed to lock map for reading: {e}")))?;
 
             if let Some(value) = guarded_map.get(&kv_key_suffix) {
-                return Ok(Some(KvPair {
-                    value: value.clone(),
+                tracing::debug!(%kv_key_suffix, "found kv value in cache");
+                return Ok(Some(mapper(KvPair {
+                    value,
                     key: kv_key_suffix,
-                }));
+                })));
             }
         }
 
         // If the value is not cached, try to populate it from KV before returning.
-        let kv_key = format!("{}/{}", kv_key_prefix, kv_key_suffix.to_string());
+        let kv_key = format!("{}/{}", kv_key_prefix, kv_key_suffix);
+
+        tracing::debug!(%kv_key, "looking up key in kv");
         let kv_store = self.kv()?;
         let builder = kv_store.get(&kv_key);
         if let Some(kv_value) = builder.json::<V>().await? {
@@ -632,29 +614,94 @@ impl<'srv> DaphneWorker<'srv> {
             .map_err(|e| Error::RustError(format!("Failed to lock map for reading: {e}")))?;
 
         if let Some(value) = guarded_map.get(kv_key_suffix.as_ref()) {
-            Ok(Some(KvPair {
-                value: value.clone(),
+            tracing::debug!(%kv_key, "found key in kv");
+            Ok(Some(mapper(KvPair {
+                value,
                 key: kv_key_suffix,
-            }))
+            })))
         } else {
             Ok(None)
         }
     }
 
-    /// Get a reference to the HPKE receiver configs, ensuring that the config indicated by
-    /// `hpke_config_id` is cached (if it exists).
-    pub(crate) async fn get_hpke_receiver_config(
+    /// In memory cache on-top of KV to avoid hitting KV API limits.
+    // NOTE: We shouldn't return guards from this function as they could be held across await points.
+    //
+    // Locks in wasm are assumed to be single threaded without concurrency:
+    // https://github.com/rust-lang/rust/blob/6f8c0557e0b73c73a8a7163a15f4a5a3feca7d5c/library/std/src/sys/unsupported/locks/rwlock.rs#L4
+    //
+    // However, with Cloudflare Workers the situation is not so straightforward.
+    //
+    // A single isolate can process multiple requests concurrently (not in parallel), so holding
+    // ReadGuards across await points is not OK.
+    //
+    // There are some official docs that refer to the behaviour:
+    // https://developers.cloudflare.com/workers/learning/how-workers-works/#distributed-execution
+    //
+    // The errors seen are:
+    //
+    //   Error: The script will never generate a response.
+    //
+    //   and
+    //
+    //   Memory access out of bounds
+    //
+    // They only seem to happen under load, likely because the Workers runtime will readily spawn
+    // additional isolates.
+    async fn kv_get_cached<'req, K, V>(
         &self,
-        hpke_receiver_kv_key: HpkeReceiverKvKey,
-    ) -> Result<Option<HpkeReceiverConfigKvPair>> {
-        self.kv_get_cached(
-            &self.isolate_state().hpke_receiver_configs,
-            KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG,
-            Cow::Owned(hpke_receiver_kv_key),
-        )
+        map: &'srv Arc<RwLock<HashMap<K, V>>>,
+        kv_key_prefix: &str,
+        kv_key_suffix: Cow<'req, K>,
+    ) -> Result<Option<KvPair<'req, K, V>>>
+    where
+        K: Clone + Eq + std::hash::Hash + Display,
+        V: Clone + for<'de> Deserialize<'de>,
+    {
+        self.kv_get_cached_mapped(map, kv_key_prefix, kv_key_suffix, |pair| KvPair {
+            value: pair.value.clone(),
+            key: pair.key,
+        })
         .await
     }
 
+    /// Get a reference to the HPKE receiver configs, ensuring that the config indicated by
+    /// `version` is cached (if it exists).
+    ///
+    /// The `mapper` let's you extract the minimum you need from the [`HpkeRecieverConfigSet`],
+    /// the goal being that you can clone as little as possible.
+    ///
+    /// If the `mapper` returns [`None`], then kv will be hit in order to make sure that we don't
+    /// miss a new value due to having a stale cache.
+    pub(crate) async fn get_hpke_receiver_config<F, R>(
+        &self,
+        version: DapVersion,
+        mut mapper: F,
+    ) -> Result<Option<R>>
+    where
+        F: FnMut(&HpkeRecieverConfigSet) -> Option<R>,
+    {
+        let cached_config = self
+            .isolate_state()
+            .hpke_receiver_configs
+            .read()
+            .unwrap()
+            .get(&version)
+            .and_then(&mut mapper);
+        match cached_config {
+            Some(config) => Ok(Some(config)),
+            None => Ok(self
+                .kv_get_cached_mapped(
+                    &self.isolate_state().hpke_receiver_configs,
+                    KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG_SET,
+                    Cow::Owned(version),
+                    |pair| mapper(pair.value),
+                )
+                .await?
+                .flatten()),
+        }
+    }
+
     /// Retrieve from KV the Leader's bearer token for the given task.
     pub(crate) async fn get_leader_bearer_token<'a>(
         &'a self,
@@ -948,6 +995,34 @@ impl<'srv> DaphneWorker<'srv> {
         }
     }
 
+    pub(crate) async fn internal_add_hpke_config(
+        &self,
+        version: DapVersion,
+        hpke: HpkeReceiverConfig,
+    ) -> Result<()> {
+        let mut configs = self
+            .get_hpke_receiver_config(version, |config_set| Some(config_set.clone()))
+            .await?
+            .unwrap_or_default();
+
+        configs.insert(hpke.config.id, hpke);
+
+        self.kv()?
+            .put(
+                &format!("{KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG_SET}/{version}"),
+                configs.clone(),
+            )?
+            .execute()
+            .await?;
+
+        self.isolate_state()
+            .hpke_receiver_configs
+            .write()
+            .unwrap()
+            .insert(version, configs);
+        Ok(())
+    }
+
     pub(crate) fn extract_version_parameter(&self, req: &Request) -> Result<DapVersion> {
         let url = req.url()?;
         let path = url.path();
@@ -1168,14 +1243,6 @@ impl<K: Clone + Eq + std::hash::Hash, V> KvPair<'_, K, V> {
     }
 }
 
-pub(crate) type HpkeReceiverConfigKvPair<'a> = KvPair<'a, HpkeReceiverKvKey, HpkeReceiverConfig>;
-
-impl AsRef<HpkeConfig> for HpkeReceiverConfigKvPair<'_> {
-    fn as_ref(&self) -> &HpkeConfig {
-        &self.value().config
-    }
-}
-
 pub(crate) type BearerTokenKvPair<'a> = KvPair<'a, TaskId, BearerToken>;
 
 impl AsRef<BearerToken> for BearerTokenKvPair<'_> {
@@ -1204,59 +1271,3 @@ pub(crate) enum DaphneWorkerDeployment {
     /// internal test API.
     Dev,
 }
-
-#[derive(Clone, Eq, Hash, PartialEq)]
-pub(crate) struct HpkeReceiverKvKey {
-    pub(crate) version: DapVersion,
-    pub(crate) hpke_config_id: u8,
-}
-
-impl HpkeReceiverKvKey {
-    fn parse_from_name(name: &str) -> Option<Self> {
-        let mut iter = name.split('/');
-
-        // Read "{KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG}/version".
-        if iter.next()? != KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG || iter.next()? != "version" {
-            return None;
-        }
-
-        // Read and parse the version.
-        let version = DapVersion::from(iter.next()?);
-        if matches!(version, DapVersion::Unknown) {
-            return None;
-        }
-
-        // Read "config_id".
-        if iter.next()? != "config_id" {
-            return None;
-        }
-
-        // Read and parse the config ID.
-        let hpke_config_id = iter.next()?.parse::<u8>().ok()?;
-
-        // Make sure there is no trailing data.
-        if iter.next().is_some() {
-            return None;
-        }
-
-        Some(HpkeReceiverKvKey {
-            version,
-            hpke_config_id,
-        })
-    }
-
-    pub(crate) fn try_from_name(name: &str) -> std::result::Result<Self, DapError> {
-        Self::parse_from_name(name)
-            .ok_or_else(|| fatal_error!(err = "encountored invalid KV name", %name))
-    }
-}
-
-impl std::fmt::Display for HpkeReceiverKvKey {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "version/{}/config_id/{}",
-            self.version, self.hpke_config_id
-        )
-    }
-}
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index 0152e9e..c59f555 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -8,10 +8,7 @@
 
 use crate::{
     auth::DaphneWorkerAuth,
-    config::{
-        BearerTokenKvPair, DapTaskConfigKvPair, DaphneWorker, HpkeReceiverConfigKvPair,
-        HpkeReceiverKvKey, KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG,
-    },
+    config::{BearerTokenKvPair, DapTaskConfigKvPair, DaphneWorker},
     durable::{
         aggregate_store::{
             DURABLE_AGGREGATE_STORE_CHECK_COLLECTED, DURABLE_AGGREGATE_STORE_GET,
@@ -50,11 +47,10 @@ use daphne::{
     constants::DapMediaType,
     error::DapAbort,
     fatal_error,
-    hpke::HpkeDecrypter,
+    hpke::{HpkeConfig, HpkeDecrypter},
     messages::{
         BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq, HpkeCiphertext,
-        HpkeConfig, PartialBatchSelector, Report, ReportId, ReportMetadata, TaskId,
-        TransitionFailure,
+        PartialBatchSelector, Report, ReportId, ReportMetadata, TaskId, TransitionFailure,
     },
     metrics::DaphneMetrics,
     roles::{early_metadata_check, DapAggregator, DapAuthorizedSender, DapHelper, DapLeader},
@@ -87,83 +83,25 @@ pub(crate) fn dap_response_to_worker(resp: DapResponse) -> Result<Response> {
 
 #[async_trait(?Send)]
 impl<'srv> HpkeDecrypter for DaphneWorker<'srv> {
-    type WrappedHpkeConfig<'a> = HpkeReceiverConfigKvPair<'a>
+    type WrappedHpkeConfig<'a> = HpkeConfig
         where Self: 'a;
 
     async fn get_hpke_config_for<'s>(
         &'s self,
         version: DapVersion,
         _task_id: Option<&TaskId>,
-    ) -> std::result::Result<HpkeReceiverConfigKvPair<'s>, DapError> {
-        let kv_store = self.kv().map_err(|e| fatal_error!(err = e))?;
-        let keys = kv_store
-            .list()
-            .limit(1)
-            .prefix(KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG.to_string())
-            .execute()
-            .await
-            .map_err(|e| fatal_error!(err = e, "failed to list hpke keys in kv"))?;
-
-        let hpke_receiver_kv_key = if keys.keys.is_empty() {
-            // Generate a new HPKE receiver config and store it in KV.
-            //
-            // For now, expect that only one KEM algorithm is supported and that only one config
-            // will be used at anyone time.
-            if self.config().global.supported_hpke_kems.len() != 1 {
-                return Err(fatal_error!(
-                    err = "The number of supported HPKE KEMs must be 1",
-                    was = self.config().global.supported_hpke_kems.len(),
-                ));
-            }
-
-            let mut hpke_config_id = None;
-            for it in self
-                .config()
-                .global
-                .gen_hpke_receiver_config_list(rand::random())
-            {
-                let hpke_receiver_config = it.expect("failed to generate HPKE receiver config");
-                if hpke_config_id.is_none() {
-                    hpke_config_id = Some(hpke_receiver_config.config.id);
-                }
-                let new_kv_config_key = format!(
-                    "{}/{}",
-                    KV_KEY_PREFIX_HPKE_RECEIVER_CONFIG,
-                    HpkeReceiverKvKey {
-                        version,
-                        hpke_config_id: hpke_receiver_config.config.id
-                    },
-                );
-
-                tracing::info!(%new_kv_config_key, "creating a new hpke config");
-                kv_store
-                    .put(&new_kv_config_key, hpke_receiver_config)
-                    .map_err(|e| fatal_error!(err = e, "failed to setup a hpke put request to kv"))?
-                    .execute()
-                    .await
-                    .map_err(|e| {
-                        fatal_error!(err = e, "failed to execute a hpke put request to kv")
-                    })?;
-            }
-
-            HpkeReceiverKvKey {
-                version,
-                hpke_config_id: hpke_config_id.unwrap(),
-            }
-        } else {
-            // Return the first HPKE receiver config in the list.
-            HpkeReceiverKvKey::try_from_name(keys.keys[0].name.as_str())?
-        };
-
-        // Fetch the indicated HPKE config from KV.
-        //
-        // TODO(cjpatton) Figure out how likely this is to fail if we had to generate a new key
-        // pair and write it to KV during this call.
-        Ok(self
-            .get_hpke_receiver_config(hpke_receiver_kv_key)
-            .await
-            .map_err(|e| fatal_error!(err = e))?
-            .ok_or_else(|| fatal_error!(err = "empty HPKE receiver config list"))?)
+    ) -> std::result::Result<Self::WrappedHpkeConfig<'s>, DapError> {
+        self.get_hpke_receiver_config(version, |receiver_config_set| {
+            // TODO: We are currently not doing key rotation so this code assumes that there is only
+            // one element in the hashmap. In the future this should change.
+            receiver_config_set
+                .iter()
+                .next()
+                .map(|(_, receiver_config)| receiver_config.config.clone())
+        })
+        .await
+        .map_err(|e| fatal_error!(err = e, "failed to get list of hpke key configs in kv"))?
+        .ok_or_else(|| fatal_error!(err = "there are no hpke configs in kv!!", %version))
     }
 
     async fn can_hpke_decrypt(
@@ -173,9 +111,11 @@ impl<'srv> HpkeDecrypter for DaphneWorker<'srv> {
     ) -> std::result::Result<bool, DapError> {
         let version = self.try_get_task_config(task_id).await?.as_ref().version;
         Ok(self
-            .get_hpke_receiver_config(HpkeReceiverKvKey {
-                version,
-                hpke_config_id: config_id,
+            .get_hpke_receiver_config(version, |config_set| {
+                config_set.get(&config_id).map(|_| {
+                    // we just need to know if it exists, so we use a Option<()> as a boolean we
+                    // test bellow with is_some
+                })
             })
             .await
             .map_err(|e| fatal_error!(err = e))?
@@ -190,23 +130,14 @@ impl<'srv> HpkeDecrypter for DaphneWorker<'srv> {
         ciphertext: &HpkeCiphertext,
     ) -> std::result::Result<Vec<u8>, DapError> {
         let version = self.try_get_task_config(task_id).await?.as_ref().version;
-        if let Some(hpke_receiver_config) = self
-            .get_hpke_receiver_config(HpkeReceiverKvKey {
-                version,
-                hpke_config_id: ciphertext.config_id,
-            })
-            .await
-            .map_err(|e| fatal_error!(err = e))?
-        {
-            Ok(hpke_receiver_config.value().decrypt(
-                info,
-                aad,
-                &ciphertext.enc,
-                &ciphertext.payload,
-            )?)
-        } else {
-            Err(DapError::Transition(TransitionFailure::HpkeUnknownConfigId))
-        }
+        self.get_hpke_receiver_config(version, |config_set| {
+            config_set
+                .get(&ciphertext.config_id)
+                .map(|config| config.decrypt(info, aad, &ciphertext.enc, &ciphertext.payload))
+        })
+        .await
+        .map_err(|e| fatal_error!(err = e))?
+        .ok_or_else(|| DapError::Transition(TransitionFailure::HpkeUnknownConfigId))?
     }
 }
 
diff --git a/daphne_worker/src/lib.rs b/daphne_worker/src/lib.rs
index b2d4181..1422233 100644
--- a/daphne_worker/src/lib.rs
+++ b/daphne_worker/src/lib.rs
@@ -163,6 +163,7 @@ use daphne::{
     auth::BearerToken,
     constants::DapMediaType,
     error::DapAbort,
+    hpke::HpkeReceiverConfig,
     messages::{CollectionJobId, Duration, TaskId, Time},
     roles::{DapAggregator, DapHelper, DapLeader},
     DapCollectJob, DapError, DapRequest, DapResponse, DapVersion,
@@ -582,6 +583,20 @@ impl DaphneWorkerRouter<'_> {
                         }))
                     },
                 )
+                .post_async(
+                    "/:version/internal/test/add_hpke_config",
+                    |mut req, ctx| async move {
+                        let daph = ctx.data.handler(&ctx.env);
+                        let hpke: HpkeReceiverConfig = req.json().await?;
+                        let version = daph.extract_version_parameter(&req)?;
+                        daph.internal_add_hpke_config(version, hpke)
+                            .instrument(info_span!("add_hpke_config"))
+                            .await?;
+                        Response::from_json(&serde_json::json!({
+                            "status": "success",
+                        }))
+                    },
+                )
         } else {
             router
         };


Test patch:
diff --git a/daphne/src/hpke_test.rs b/daphne/src/hpke_test.rs
index 300ccc4..f422aa4 100644
--- a/daphne/src/hpke_test.rs
+++ b/daphne/src/hpke_test.rs
@@ -1,8 +1,7 @@
 // Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use crate::hpke::HpkeReceiverConfig;
-use crate::messages::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId};
+use crate::hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig};
 use hpke_rs::{Hpke, HpkePrivateKey, HpkePublicKey, Mode};
 use hpke_rs_crypto::types::{AeadAlgorithm, KdfAlgorithm, KemAlgorithm};
 use hpke_rs_rust_crypto::HpkeRustCrypto as ImplHpkeCrypto;
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index 95dcc93..2141386 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -6,12 +6,12 @@ use crate::{
     async_test_versions,
     auth::BearerToken,
     constants::DapMediaType,
-    hpke::{HpkeDecrypter, HpkeReceiverConfig},
+    hpke::{HpkeDecrypter, HpkeKemId, HpkeReceiverConfig},
     messages::{
         taskprov, AggregateShareReq, AggregationJobContinueReq, AggregationJobInitReq,
         AggregationJobResp, BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq,
-        Extension, HpkeKemId, Interval, PartialBatchSelector, Query, Report, ReportId,
-        ReportMetadata, ReportShare, TaskId, Time, Transition, TransitionFailure, TransitionVar,
+        Extension, Interval, PartialBatchSelector, Query, Report, ReportId, ReportMetadata,
+        ReportShare, TaskId, Time, Transition, TransitionFailure, TransitionVar,
     },
     metrics::DaphneMetrics,
     roles::{early_metadata_check, DapAggregator, DapAuthorizedSender, DapHelper, DapLeader},
diff --git a/daphne/src/taskprov_test.rs b/daphne/src/taskprov_test.rs
index 2d5c120..8445590 100644
--- a/daphne/src/taskprov_test.rs
+++ b/daphne/src/taskprov_test.rs
@@ -5,9 +5,9 @@ use prio::codec::ParameterizedEncode;
 
 use crate::{
     auth::BearerToken,
-    hpke::HpkeReceiverConfig,
+    hpke::{HpkeKemId, HpkeReceiverConfig},
+    messages::taskprov::VdafType,
     messages::{encode_base64url, taskprov::*, Extension, ReportId, ReportMetadata, TaskId},
-    messages::{taskprov::VdafType, HpkeKemId},
     taskprov::{
         compute_task_id, compute_vdaf_verify_key, resolve_advertised_task_config, TaskprovVersion,
     },
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 01281bf..e198ad4 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -8,11 +8,11 @@ use crate::{
     auth::{BearerToken, BearerTokenProvider},
     constants::DapMediaType,
     fatal_error,
-    hpke::{HpkeDecrypter, HpkeReceiverConfig},
+    hpke::{HpkeConfig, HpkeDecrypter, HpkeReceiverConfig},
     messages::{
         AggregationJobId, BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq,
-        Draft02AggregationJobId, HpkeCiphertext, HpkeConfig, PartialBatchSelector, Report,
-        ReportId, ReportMetadata, TaskId, Time, TransitionFailure,
+        Draft02AggregationJobId, HpkeCiphertext, PartialBatchSelector, Report, ReportId,
+        ReportMetadata, TaskId, Time, TransitionFailure,
     },
     metrics::DaphneMetrics,
     roles::{DapAggregator, DapAuthorizedSender, DapHelper, DapLeader},
diff --git a/daphne/src/vdaf/mod_test.rs b/daphne/src/vdaf/mod_test.rs
index 51c6972..3ec6b07 100644
--- a/daphne/src/vdaf/mod_test.rs
+++ b/daphne/src/vdaf/mod_test.rs
@@ -5,12 +5,11 @@ use crate::{
     assert_metrics_include, assert_metrics_include_auxiliary_function, async_test_version,
     async_test_versions,
     error::DapAbort,
-    hpke::HpkeReceiverConfig,
+    hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig},
     messages::{
         AggregationJobContinueReq, AggregationJobInitReq, AggregationJobResp, BatchSelector,
-        HpkeAeadId, HpkeCiphertext, HpkeConfig, HpkeKdfId, HpkeKemId, Interval,
-        PartialBatchSelector, Report, ReportId, ReportShare, TaskId, Time, Transition,
-        TransitionFailure, TransitionVar,
+        HpkeCiphertext, Interval, PartialBatchSelector, Report, ReportId, ReportShare, TaskId,
+        Time, Transition, TransitionFailure, TransitionVar,
     },
     metrics::DaphneMetrics,
     test_version, test_versions, DapAggregateResult, DapAggregateShare, DapError, DapHelperState,
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index 6e345f9..a120b19 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -6,10 +6,9 @@
 use assert_matches::assert_matches;
 use daphne::{
     constants::DapMediaType,
-    hpke::HpkeReceiverConfig,
+    hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        encode_base64url, BatchId, CollectionJobId, Duration, HpkeAeadId, HpkeConfig,
-        HpkeConfigList, HpkeKdfId, HpkeKemId, Interval, TaskId,
+        encode_base64url, BatchId, CollectionJobId, Duration, HpkeConfigList, Interval, TaskId,
     },
     taskprov::TaskprovVersion,
     DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
@@ -31,7 +30,7 @@ pub(crate) const MAX_BATCH_SIZE: u64 = 12;
 pub(crate) const TIME_PRECISION: Duration = 3600; // seconds
 
 #[derive(Deserialize)]
-struct InternalTestAddTaskResult {
+struct InternalTestCommandResult {
     status: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     error: Option<String>,
@@ -203,7 +202,7 @@ impl TestRunner {
             "task_expiration": t.task_config.expiration,
         });
         let add_task_path = format!("{}/internal/test/add_task", version.as_ref());
-        let res: InternalTestAddTaskResult = t
+        let res: InternalTestCommandResult = t
             .leader_post_internal(&add_task_path, &leader_add_task_cmd)
             .await;
         assert_eq!(
@@ -228,7 +227,7 @@ impl TestRunner {
             "collector_hpke_config": collector_hpke_config_base64url.clone(),
             "task_expiration": t.task_config.expiration,
         });
-        let res: InternalTestAddTaskResult = t
+        let res: InternalTestCommandResult = t
             .helper_post_internal(&add_task_path, &helper_add_task_cmd)
             .await;
         assert_eq!(
@@ -237,6 +236,36 @@ impl TestRunner {
             res.status, res.error
         );
 
+        let gen_config = || {
+            HpkeReceiverConfig::gen(random(), HpkeKemId::X25519HkdfSha256)
+                .expect("failed to generate receiver config")
+        };
+        let res: InternalTestCommandResult = t
+            .helper_post_internal(
+                &format!("{version}/internal/test/add_hpke_config"),
+                &gen_config(),
+            )
+            .await;
+
+        assert_eq!(
+            res.status, "success",
+            "response status: {}, error {:?}",
+            res.status, res.error
+        );
+
+        let res: InternalTestCommandResult = t
+            .leader_post_internal(
+                &format!("{version}/internal/test/add_hpke_config"),
+                &gen_config(),
+            )
+            .await;
+
+        assert_eq!(
+            res.status, "success",
+            "response status: {}, error {:?}",
+            res.status, res.error
+        );
+
         t
     }