Rename DurableRequestPayload to CapnprotoPayload

Author: mendess

crates/daphne-server/src/storage_proxy_connection/mod.rs

@@ -6,9 +6,9 @@ pub(crate) mod kv;
 use std::fmt::Debug;
 
 use axum::http::StatusCode;
-use daphne_service_utils::durable_requests::{
-    bindings::{DurableMethod, DurableRequestPayload, DurableRequestPayloadExt},
-    DurableRequest, ObjectIdFrom, DO_PATH_PREFIX,
+use daphne_service_utils::{
+    capnproto_payload::{CapnprotoPayload, CapnprotoPayloadExt as _},
+    durable_requests::{bindings::DurableMethod, DurableRequest, ObjectIdFrom, DO_PATH_PREFIX},
 };
 use serde::de::DeserializeOwned;
 
@@ -95,7 +95,7 @@ impl<'d, B: DurableMethod + Debug, P: AsRef<[u8]>> RequestBuilder<'d, B, P> {
 }
 
 impl<'d, B: DurableMethod> RequestBuilder<'d, B, [u8; 0]> {
-    pub fn encode<T: DurableRequestPayload>(self, payload: &T) -> RequestBuilder<'d, B, Vec<u8>> {
+    pub fn encode<T: CapnprotoPayload>(self, payload: &T) -> RequestBuilder<'d, B, Vec<u8>> {
         self.with_body(payload.encode_to_bytes().unwrap())
     }
 

crates/daphne-service-utils/src/capnproto_payload.rs

@@ -0,0 +1,44 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+pub trait CapnprotoPayload {
+    fn decode_from_reader(
+        reader: capnp::message::Reader<capnp::serialize::OwnedSegments>,
+    ) -> capnp::Result<Self>
+    where
+        Self: Sized;
+
+    fn encode_to_builder(&self) -> capnp::message::Builder<capnp::message::HeapAllocator>;
+}
+
+pub trait CapnprotoPayloadExt {
+    fn decode_from_bytes(bytes: &[u8]) -> capnp::Result<Self>
+    where
+        Self: Sized;
+    fn encode_to_bytes(&self) -> capnp::Result<Vec<u8>>;
+}
+
+impl<T> CapnprotoPayloadExt for T
+where
+    T: CapnprotoPayload,
+{
+    fn encode_to_bytes(&self) -> capnp::Result<Vec<u8>> {
+        let mut buf = Vec::new();
+        let message = self.encode_to_builder();
+        capnp::serialize_packed::write_message(&mut buf, &message)?;
+        Ok(buf)
+    }
+
+    fn decode_from_bytes(bytes: &[u8]) -> capnp::Result<Self>
+    where
+        Self: Sized,
+    {
+        let mut cursor = std::io::Cursor::new(bytes);
+        let reader = capnp::serialize_packed::read_message(
+            &mut cursor,
+            capnp::message::ReaderOptions::new(),
+        )?;
+
+        T::decode_from_reader(reader)
+    }
+}

crates/daphne-service-utils/src/durable_requests/bindings/aggregate_store.rs

@@ -11,11 +11,11 @@ use daphne::{
 use serde::{Deserialize, Serialize};
 
 use crate::{
+    capnproto_payload::CapnprotoPayload,
     durable_request_capnp::{aggregate_store_merge_req, dap_aggregate_share},
     durable_requests::ObjectIdFrom,
 };
 
-use super::DurableRequestPayload;
 use prio::{
     field::{FieldElement, FieldError},
     vdaf::AggregateShare,
@@ -68,7 +68,7 @@ pub struct AggregateStoreMergeOptions {
     pub skip_replay_protection: bool,
 }
 
-impl DurableRequestPayload for AggregateStoreMergeReq {
+impl CapnprotoPayload for AggregateStoreMergeReq {
     fn encode_to_builder(&self) -> capnp::message::Builder<capnp::message::HeapAllocator> {
         let Self {
             contained_reports,
@@ -285,7 +285,7 @@ mod test {
     };
     use rand::{thread_rng, Rng};
 
-    use crate::durable_requests::bindings::DurableRequestPayloadExt;
+    use crate::capnproto_payload::CapnprotoPayloadExt as _;
 
     use super::*;
 

crates/daphne-service-utils/src/durable_requests/bindings/mod.rs

@@ -37,48 +37,6 @@ pub trait DurableMethod {
     fn name(params: Self::NameParameters<'_>) -> ObjectIdFrom;
 }
 
-pub trait DurableRequestPayload {
-    fn decode_from_reader(
-        reader: capnp::message::Reader<capnp::serialize::OwnedSegments>,
-    ) -> capnp::Result<Self>
-    where
-        Self: Sized;
-
-    fn encode_to_builder(&self) -> capnp::message::Builder<capnp::message::HeapAllocator>;
-}
-
-pub trait DurableRequestPayloadExt {
-    fn decode_from_bytes(bytes: &[u8]) -> capnp::Result<Self>
-    where
-        Self: Sized;
-    fn encode_to_bytes(&self) -> capnp::Result<Vec<u8>>;
-}
-
-impl<T> DurableRequestPayloadExt for T
-where
-    T: DurableRequestPayload,
-{
-    fn encode_to_bytes(&self) -> capnp::Result<Vec<u8>> {
-        let mut buf = Vec::new();
-        let message = self.encode_to_builder();
-        capnp::serialize_packed::write_message(&mut buf, &message)?;
-        Ok(buf)
-    }
-
-    fn decode_from_bytes(bytes: &[u8]) -> capnp::Result<Self>
-    where
-        Self: Sized,
-    {
-        let mut cursor = std::io::Cursor::new(bytes);
-        let reader = capnp::serialize_packed::read_message(
-            &mut cursor,
-            capnp::message::ReaderOptions::new(),
-        )?;
-
-        T::decode_from_reader(reader)
-    }
-}
-
 macro_rules! define_do_binding {
     (
         const BINDING = $binding:literal;

crates/daphne-service-utils/src/lib.rs

@@ -5,6 +5,8 @@
 
 pub mod bearer_token;
 #[cfg(feature = "durable_requests")]
+pub mod capnproto_payload;
+#[cfg(feature = "durable_requests")]
 pub mod durable_requests;
 pub mod http_headers;
 #[cfg(feature = "test-utils")]

crates/daphne-worker/src/durable/mod.rs

@@ -24,8 +24,9 @@ pub(crate) mod aggregate_store;
 pub(crate) mod test_state_cleaner;
 
 use crate::tracing_utils::shorten_paths;
-use daphne_service_utils::durable_requests::bindings::{
-    DurableMethod, DurableRequestPayload, DurableRequestPayloadExt,
+use daphne_service_utils::{
+    capnproto_payload::{CapnprotoPayload, CapnprotoPayloadExt},
+    durable_requests::bindings::DurableMethod,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info_span;
@@ -209,7 +210,7 @@ pub(crate) async fn state_set_if_not_exists<T: for<'a> Deserialize<'a> + Seriali
 
 async fn req_parse<T>(req: &mut Request) -> Result<T>
 where
-    T: DurableRequestPayload,
+    T: CapnprotoPayload,
 {
     T::decode_from_bytes(&req.bytes().await?).map_err(|e| Error::RustError(e.to_string()))
 }