diff --git a/src/common/etcd/impl/hostports_observer.h b/src/common/etcd/impl/hostports_observer.h
index ec6829368..3bcde61c7 100644
--- a/src/common/etcd/impl/hostports_observer.h
+++ b/src/common/etcd/impl/hostports_observer.h
@@ -126,7 +126,6 @@ class hostports_observer : public subscriber_observer {
     service_factory<service_interface> m_service_factory;
     std::vector<std::reference_wrapper<service_observer<service_interface>>>
         m_observers;
-    // TODO: Modify this to vector and get num_storages on the constructor
     std::map<std::size_t, std::shared_ptr<service_interface>> m_clients;
     std::atomic<std::size_t> m_client_count{0};
 };
diff --git a/src/common/telemetry/trace/awaitable_operators.h b/src/common/telemetry/trace/awaitable_operators.h
new file mode 100644
index 000000000..c6a0ff619
--- /dev/null
+++ b/src/common/telemetry/trace/awaitable_operators.h
@@ -0,0 +1,494 @@
+#pragma once
+
+#include <common/telemetry/trace/trace_asio.h>
+
+#include <boost/asio/experimental/parallel_group.hpp>
+
+namespace boost {
+namespace asio {
+namespace experimental {
+namespace awaitable_operators {
+namespace detail {
+
+template <typename T, typename Executor>
+traced_awaitable<T, Executor>
+awaitable_wrap(traced_awaitable<T, Executor> a,
+               constraint_t<is_constructible<T>::value>* = 0) {
+    return a;
+}
+
+template <typename T, typename Executor>
+traced_awaitable<std::optional<T>, Executor>
+awaitable_wrap(traced_awaitable<T, Executor> a,
+               constraint_t<!is_constructible<T>::value>* = 0) {
+    co_return std::optional<T>(co_await std::move(a));
+}
+
+template <typename T>
+T& awaitable_unwrap(conditional_t<true, T, void>& r,
+                    constraint_t<is_constructible<T>::value>* = 0) {
+    return r;
+}
+
+template <typename T>
+T& awaitable_unwrap(std::optional<conditional_t<true, T, void>>& r,
+                    constraint_t<!is_constructible<T>::value>* = 0) {
+    return *r;
+}
+
+} // namespace detail
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename Executor>
+traced_awaitable<void, Executor>
+operator&&(traced_awaitable<void, Executor> t,
+           traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, ex1] =
+        co_await make_parallel_group(
+            co_spawn(ex, std::move(t.continue_trace(context)), deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return;
+}
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename U, typename Executor>
+traced_awaitable<U, Executor> operator&&(traced_awaitable<void, Executor> t,
+                                         traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(ex, std::move(t.continue_trace(context)), deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return std::move(detail::awaitable_unwrap<U>(r1));
+}
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename T, typename Executor>
+traced_awaitable<T, Executor> operator&&(traced_awaitable<T, Executor> t,
+                                         traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return std::move(detail::awaitable_unwrap<T>(r0));
+}
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename T, typename U, typename Executor>
+traced_awaitable<std::tuple<T, U>, Executor>
+operator&&(traced_awaitable<T, Executor> t, traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return std::make_tuple(std::move(detail::awaitable_unwrap<T>(r0)),
+                              std::move(detail::awaitable_unwrap<U>(r1)));
+}
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename... T, typename Executor>
+traced_awaitable<std::tuple<T..., std::monostate>, Executor>
+operator&&(traced_awaitable<std::tuple<T...>, Executor> t,
+           traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return std::move(detail::awaitable_unwrap<std::tuple<T...>>(r0));
+}
+
+/// Wait for both operations to succeed.
+/**
+ * If one operations fails, the other is cancelled as the AND-condition can no
+ * longer be satisfied.
+ */
+template <typename... T, typename U, typename Executor>
+traced_awaitable<std::tuple<T..., U>, Executor>
+operator&&(traced_awaitable<std::tuple<T...>, Executor> t,
+           traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_error(), deferred);
+
+    if (ex0 && ex1)
+        throw multiple_exceptions(ex0);
+    if (ex0)
+        std::rethrow_exception(ex0);
+    if (ex1)
+        std::rethrow_exception(ex1);
+    co_return std::tuple_cat(
+        std::move(detail::awaitable_unwrap<std::tuple<T...>>(r0)),
+        std::make_tuple(std::move(detail::awaitable_unwrap<U>(r1))));
+}
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename Executor>
+traced_awaitable<std::variant<std::monostate, std::monostate>, Executor>
+operator||(traced_awaitable<void, Executor> t,
+           traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, ex1] =
+        co_await make_parallel_group(
+            co_spawn(ex, std::move(t.continue_trace(context)), deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return std::variant<std::monostate, std::monostate>{
+                std::in_place_index<0>};
+        if (!ex1)
+            co_return std::variant<std::monostate, std::monostate>{
+                std::in_place_index<1>};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<std::monostate, std::monostate>{
+                std::in_place_index<1>};
+        if (!ex0)
+            co_return std::variant<std::monostate, std::monostate>{
+                std::in_place_index<0>};
+        throw multiple_exceptions(ex1);
+    }
+}
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename U, typename Executor>
+traced_awaitable<std::variant<std::monostate, U>, Executor>
+operator||(traced_awaitable<void, Executor> t,
+           traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(ex, std::move(t.continue_trace(context)), deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return std::variant<std::monostate, U>{std::in_place_index<0>};
+        if (!ex1)
+            co_return std::variant<std::monostate, U>{
+                std::in_place_index<1>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<std::monostate, U>{
+                std::in_place_index<1>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        if (!ex0)
+            co_return std::variant<std::monostate, U>{std::in_place_index<0>};
+        throw multiple_exceptions(ex1);
+    }
+}
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename T, typename Executor>
+traced_awaitable<std::variant<T, std::monostate>, Executor>
+operator||(traced_awaitable<T, Executor> t,
+           traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return std::variant<T, std::monostate>{
+                std::in_place_index<0>,
+                std::move(detail::awaitable_unwrap<T>(r0))};
+        if (!ex1)
+            co_return std::variant<T, std::monostate>{std::in_place_index<1>};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<T, std::monostate>{std::in_place_index<1>};
+        if (!ex0)
+            co_return std::variant<T, std::monostate>{
+                std::in_place_index<0>,
+                std::move(detail::awaitable_unwrap<T>(r0))};
+        throw multiple_exceptions(ex1);
+    }
+}
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename T, typename U, typename Executor>
+traced_awaitable<std::variant<T, U>, Executor>
+operator||(traced_awaitable<T, Executor> t, traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return std::variant<T, U>{
+                std::in_place_index<0>,
+                std::move(detail::awaitable_unwrap<T>(r0))};
+        if (!ex1)
+            co_return std::variant<T, U>{
+                std::in_place_index<1>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<T, U>{
+                std::in_place_index<1>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        if (!ex0)
+            co_return std::variant<T, U>{
+                std::in_place_index<0>,
+                std::move(detail::awaitable_unwrap<T>(r0))};
+        throw multiple_exceptions(ex1);
+    }
+}
+
+namespace detail {
+
+template <typename... T> struct widen_variant {
+    template <std::size_t I, typename SourceVariant>
+    static std::variant<T...> call(SourceVariant& source) {
+        if (source.index() == I)
+            return std::variant<T...>{std::in_place_index<I>,
+                                      std::move(std::get<I>(source))};
+        else if constexpr (I + 1 < std::variant_size_v<SourceVariant>)
+            return call<I + 1>(source);
+        else
+            throw std::logic_error("empty variant");
+    }
+};
+
+} // namespace detail
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename... T, typename Executor>
+traced_awaitable<std::variant<T..., std::monostate>, Executor>
+operator||(traced_awaitable<std::variant<T...>, Executor> t,
+           traced_awaitable<void, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(ex, std::move(u.continue_trace(context)), deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    using widen = detail::widen_variant<T..., std::monostate>;
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return widen::template call<0>(
+                detail::awaitable_unwrap<std::variant<T...>>(r0));
+        if (!ex1)
+            co_return std::variant<T..., std::monostate>{
+                std::in_place_index<sizeof...(T)>};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<T..., std::monostate>{
+                std::in_place_index<sizeof...(T)>};
+        if (!ex0)
+            co_return widen::template call<0>(
+                detail::awaitable_unwrap<std::variant<T...>>(r0));
+        throw multiple_exceptions(ex1);
+    }
+}
+
+/// Wait for one operation to succeed.
+/**
+ * If one operations succeeds, the other is cancelled as the OR-condition is
+ * already satisfied.
+ */
+template <typename... T, typename U, typename Executor>
+traced_awaitable<std::variant<T..., U>, Executor>
+operator||(traced_awaitable<std::variant<T...>, Executor> t,
+           traced_awaitable<U, Executor> u) {
+    auto ex = co_await this_coro::executor;
+    auto context = co_await this_coro::context;
+
+    auto [order, ex0, r0, ex1, r1] =
+        co_await make_parallel_group(
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(t.continue_trace(context))),
+                deferred),
+            co_spawn(
+                ex,
+                detail::awaitable_wrap(std::move(u.continue_trace(context))),
+                deferred))
+            .async_wait(wait_for_one_success(), deferred);
+
+    using widen = detail::widen_variant<T..., U>;
+    if (order[0] == 0) {
+        if (!ex0)
+            co_return widen::template call<0>(
+                detail::awaitable_unwrap<std::variant<T...>>(r0));
+        if (!ex1)
+            co_return std::variant<T..., U>{
+                std::in_place_index<sizeof...(T)>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        throw multiple_exceptions(ex0);
+    } else {
+        if (!ex1)
+            co_return std::variant<T..., U>{
+                std::in_place_index<sizeof...(T)>,
+                std::move(detail::awaitable_unwrap<U>(r1))};
+        if (!ex0)
+            co_return widen::template call<0>(
+                detail::awaitable_unwrap<std::variant<T...>>(r0));
+        throw multiple_exceptions(ex1);
+    }
+}
+
+} // namespace awaitable_operators
+} // namespace experimental
+} // namespace asio
+} // namespace boost
diff --git a/src/common/types/common_types.h b/src/common/types/common_types.h
index 3d87ea1e9..46432833b 100644
--- a/src/common/types/common_types.h
+++ b/src/common/types/common_types.h
@@ -37,6 +37,21 @@ struct refcount_t {
 using utc_time = std::chrono::time_point<std::chrono::system_clock>;
 
 template <typename T> using coro = boost::asio::traced_awaitable<T>;
+inline coro<void> async_noop() { co_return; };
+
+template <typename T> struct is_boost_awaitable : std::false_type {};
+
+template <typename T>
+struct is_boost_awaitable<boost::asio::awaitable<T>> : std::true_type {};
+
+template <typename T>
+constexpr bool is_boost_awaitable_v = is_boost_awaitable<T>::value;
+
+template <typename Awaitable>
+requires is_boost_awaitable_v<std::decay_t<Awaitable>>
+inline coro<void> async_wrap(Awaitable&& v) {
+    co_await std::move(v);
+};
 
 inline thread_local opentelemetry::context::Context THREAD_LOCAL_CONTEXT;
 
diff --git a/src/common/utils/pointer_traits.h b/src/common/utils/pointer_traits.h
index f1b1dcdcb..a0504c113 100644
--- a/src/common/utils/pointer_traits.h
+++ b/src/common/utils/pointer_traits.h
@@ -8,28 +8,11 @@ namespace uh::cluster {
 
 struct pointer_traits {
 
-    /**
-     * TODO: Let's remove this: The storage layer will receive the storage
-     * address space pointer, so they do not need to call this function
-     * themselves
-     *
-     * The data store internal pointer is the low number of uint128_t
-     *
-     * @param global_pointer
-     * @return internal data store pointer
-     */
     constexpr static const inline std::size_t group_id_bit_offset = 32 + 64;
 
     struct rr {
         constexpr static const inline std::size_t storage_id_bit_offset = 64;
 
-        inline static std::pair<std::size_t, storage_pointer>
-        get_storage_pointer(pointer global_pointer) {
-            std::size_t storage_id = (global_pointer >> 64) & 0xFFFFFFFF;
-            std::size_t storage_ptr = static_cast<std::size_t>(global_pointer);
-            return {storage_id, storage_ptr};
-        }
-
         /**
          * @param pointer
          * @param storage_id
@@ -42,6 +25,13 @@ struct pointer_traits {
                    (static_cast<pointer>(storage_id) << storage_id_bit_offset) |
                    storage_pointer;
         }
+
+        inline static std::pair<std::size_t, storage_pointer>
+        get_storage_pointer(pointer global_pointer) {
+            std::size_t storage_id = (global_pointer >> 64) & 0xFFFFFFFF;
+            std::size_t storage_ptr = static_cast<std::size_t>(global_pointer);
+            return {storage_id, storage_ptr};
+        }
     };
 
     struct ec {
diff --git a/src/entrypoint/commands/iam/delete_access_key.cpp b/src/entrypoint/commands/iam/delete_access_key.cpp
index 2f92a1242..ed453f9c0 100644
--- a/src/entrypoint/commands/iam/delete_access_key.cpp
+++ b/src/entrypoint/commands/iam/delete_access_key.cpp
@@ -17,7 +17,6 @@ coro<ep::http::response> delete_access_key::handle(ep::http::request& req) {
     if (username) {
         auto user = co_await m_users.find_by_key(*access_key);
         if (user.name != *username) {
-            // TODO: how?
             throw command_exception(
                 ep::http::status::conflict, "UserNameMismatch",
                 "AWS IAM implements sophisticated organizations/roles "
diff --git a/src/proxy/asio.h b/src/proxy/asio.h
new file mode 100644
index 000000000..50a70f5a7
--- /dev/null
+++ b/src/proxy/asio.h
@@ -0,0 +1,35 @@
+#pragma once
+
+#include <boost/asio/buffer.hpp>
+#include <span>
+
+namespace boost::asio {
+
+inline std::span<const char> get_span(boost::asio::const_buffer buffer) {
+    return {static_cast<const char*>(buffer.data()), buffer.size()};
+}
+
+inline std::span<char> get_span(boost::asio::mutable_buffer buffer) {
+    return {static_cast<char*>(buffer.data()), buffer.size()};
+}
+
+} // namespace boost::asio
+
+namespace std {
+
+template <typename T>
+inline std::span<const char> get_span(const std::vector<T>& v) {
+    return {reinterpret_cast<const char*>(v.data()), v.size() * sizeof(T)};
+}
+
+template <typename T> inline std::span<char> get_span(std::vector<T>& v) {
+    return {reinterpret_cast<char*>(v.data()), v.size() * sizeof(T)};
+}
+
+inline std::span<const char> get_span(const std::string& s) {
+    return {s.data(), s.size()};
+}
+
+inline std::span<char> get_span(std::string& s) { return {s.data(), s.size()}; }
+
+} // namespace std
diff --git a/src/proxy/cache/asio.h b/src/proxy/cache/asio.h
deleted file mode 100644
index fb3167dc2..000000000
--- a/src/proxy/cache/asio.h
+++ /dev/null
@@ -1,90 +0,0 @@
-#pragma once
-
-#include <common/types/common_types.h>
-#include <concepts>
-#include <entrypoint/http/body.h>
-#include <entrypoint/http/stream.h>
-
-namespace uh::cluster::proxy::cache {
-
-template <typename T>
-concept ReaderBodyType = requires(T r, std::span<const char> sv) {
-    { r.put(sv) } -> std::same_as<coro<std::size_t>>;
-};
-
-template <typename T>
-concept WriterBodyType = requires(T w) {
-    { w.get() } -> std::same_as<coro<std::span<const char>>>;
-};
-
-template <typename T>
-concept BodyType = requires {
-    typename T::writer;
-    typename T::reader;
-    requires WriterBodyType<typename T::writer>;
-    requires ReaderBodyType<typename T::reader>;
-};
-
-template <BodyType Body> typename Body::reader make_reader(Body& b) {
-    return typename Body::reader(b);
-}
-
-template <BodyType Body> typename Body::writer make_writer(Body& b) {
-    return typename Body::writer(b);
-}
-
-/*
- * async_read gets stream, body and size for it's input.
- *
- * size can be replaced with parser implementation
- */
-template <typename T>
-coro<void> async_read(ep::http::stream& s, T& t, std::size_t size) {
-    auto&& reader = [&]() -> auto&& {
-        if constexpr (BodyType<T>) {
-            return make_reader(t);
-        } else if constexpr (ReaderBodyType<T>) {
-            return t;
-        } else {
-            static_assert(BodyType<T> || ReaderBodyType<T>,
-                          "T must satisfy BodyType or ReaderBodyType");
-        }
-    }();
-    while (size > 0) {
-        auto sv = co_await s.read(size);
-        if (sv.empty())
-            break;
-        auto read = co_await reader.put(sv);
-        if (read != sv.size()) {
-            throw std::runtime_error(
-                "reader_body put() returned unexpected size");
-        }
-        co_await s.consume();
-        size -= sv.size();
-    }
-}
-
-/*
- * It consumes automatically
- */
-template <typename T> coro<void> async_write(ep::http::stream& s, T& t) {
-    auto&& writer = [&]() -> auto&& {
-        if constexpr (BodyType<T>) {
-            return make_writer(t);
-        } else if constexpr (WriterBodyType<T>) {
-            return t;
-        } else {
-            static_assert(BodyType<T> || WriterBodyType<T>,
-                          "T must satisfy BodyType or WriterBodyType");
-        }
-    }();
-
-    while (true) {
-        std::span<const char> data = co_await writer.get();
-        if (data.empty())
-            break;
-        co_await s.write(data);
-    }
-}
-
-} // namespace uh::cluster::proxy::cache
diff --git a/src/proxy/cache/disk/body.h b/src/proxy/cache/disk/body.h
deleted file mode 100644
index 2dc14092b..000000000
--- a/src/proxy/cache/disk/body.h
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * HTTP writer/reader bodies which supports get/put API only
- */
-#pragma once
-
-#include <proxy/cache/asio.h>
-
-#include <proxy/cache/disk/object.h>
-#include <proxy/cache/disk/utils.h>
-
-#include <common/crypto/hash.h>
-#include <common/types/common_types.h>
-#include <common/utils/strings.h>
-#include <entrypoint/http/body.h>
-#include <entrypoint/http/stream.h>
-#include <storage/interfaces/data_view.h>
-
-namespace uh::cluster::proxy::cache::disk {
-
-class reader_body {
-public:
-    reader_body(storage::data_view& writer)
-        : m_storage{writer},
-          m_addr{} {}
-
-    coro<std::size_t> put(std::span<const char> sv) {
-
-        auto addr = co_await m_storage.write(sv, {0});
-        m_hash.consume(sv);
-        m_addr.append(addr);
-        co_return addr.data_size();
-    }
-
-    /*
-     * Moves and returns the internal resource.
-     * May only be called once; further calls will return an empty or invalid
-     * value.
-     */
-    object_handle get_object_handle() {
-        // TODO: set etag with `to_hex(m_hash.finalize())`
-        return object_handle(std::move(m_addr));
-    }
-
-private:
-    storage::data_view& m_storage;
-
-    md5 m_hash;
-    address m_addr;
-};
-
-class writer_body {
-public:
-    writer_body(storage::data_view& storage,
-                std::shared_ptr<object_handle> objh,
-                std::size_t buffer_size = 32 * MEBI_BYTE)
-        : m_storage(storage),
-          m_objh{std::move(objh)},
-          m_buffer(buffer_size) {}
-
-    coro<std::span<const char>> get() {
-        std::size_t read_size = 0;
-
-        address partial_addr;
-        while (m_addr_index < m_objh->get_address().size() &&
-               read_size < m_buffer.size()) {
-
-            auto frag = m_objh->get_address().get(m_addr_index);
-            if (m_frag_offset > 0) {
-                frag.pointer += m_frag_offset;
-                frag.size -= m_frag_offset;
-            }
-            if (frag.size + read_size > m_buffer.size()) {
-                auto remains = m_buffer.size() - read_size;
-                m_frag_offset += remains;
-                frag.size = remains;
-                partial_addr.push(frag);
-            } else {
-                m_frag_offset = 0;
-                partial_addr.push(frag);
-                m_addr_index++;
-            }
-
-            read_size += frag.size;
-        }
-
-        if (read_size > 0) {
-            co_await m_storage.read_address(partial_addr,
-                                            {m_buffer.data(), read_size});
-        }
-        co_return std::span<const char>{m_buffer.data(), read_size};
-    }
-
-private:
-    storage::data_view& m_storage;
-    std::shared_ptr<object_handle> m_objh;
-
-    std::vector<char> m_buffer;
-
-    std::size_t m_addr_index = 0;
-    std::size_t m_frag_offset = 0;
-};
-
-} // namespace uh::cluster::proxy::cache::disk
diff --git a/src/proxy/cache/disk/deletion_queue.h b/src/proxy/cache/disk/deletion_queue.h
index d42b7cfc5..facd79e06 100644
--- a/src/proxy/cache/disk/deletion_queue.h
+++ b/src/proxy/cache/disk/deletion_queue.h
@@ -13,7 +13,6 @@ namespace uh::cluster::proxy::cache::disk {
 template <typename Key, EntryType Entry> class deletion_queue {
 public:
     void push(std::shared_ptr<Entry> e) {
-        // TODO: Implement this method
         std::unique_lock lock(m_mutex);
         m_queue.push(e);
         m_current_size += e->data_size();
diff --git a/src/proxy/cache/disk/disk_io.h b/src/proxy/cache/disk/disk_io.h
new file mode 100644
index 000000000..e03eb9796
--- /dev/null
+++ b/src/proxy/cache/disk/disk_io.h
@@ -0,0 +1,108 @@
+/*
+ * Sync/source for disk, which supports put/get API
+ */
+#pragma once
+
+#include <proxy/cache/disk/object.h>
+#include <proxy/cache/disk/utils.h>
+
+#include <common/crypto/hash.h>
+#include <common/types/common_types.h>
+#include <common/utils/strings.h>
+#include <entrypoint/http/body.h>
+#include <entrypoint/http/stream.h>
+#include <storage/interfaces/data_view.h>
+
+namespace uh::cluster::proxy::cache::disk {
+
+class disk_sink {
+public:
+    disk_sink(storage::data_view& writer)
+        : m_storage{writer},
+          m_addr{} {}
+
+    disk_sink(const disk_sink&) = delete;
+    disk_sink& operator=(const disk_sink&) = delete;
+    disk_sink(disk_sink&&) = default;
+    disk_sink& operator=(disk_sink&&) = default;
+
+    coro<void> put(std::span<const char> sv) {
+        if (sv.size() == 0) {
+            co_return;
+        }
+        auto addr = co_await m_storage.get().write(sv, {0});
+        m_addr.append(addr);
+    }
+
+    void set_header_size(std::size_t size) { m_header_size = size; }
+    /*
+     * Moves and returns the internal resource.
+     * May only be called once; further calls will return an empty or invalid
+     * value.
+     */
+    object_handle get_object_handle() {
+        return object_handle(std::move(m_addr), m_header_size);
+    }
+
+private:
+    std::reference_wrapper<storage::data_view> m_storage;
+
+    address m_addr;
+    std::size_t m_header_size{0};
+};
+
+class disk_source {
+public:
+    disk_source(storage::data_view& storage,
+                std::shared_ptr<object_handle> objh)
+        : m_storage(storage),
+          m_objh{std::move(objh)} {}
+
+    disk_source(const disk_source&) = delete;
+    disk_source& operator=(const disk_source&) = delete;
+    disk_source(disk_source&&) = default;
+    disk_source& operator=(disk_source&&) = default;
+
+    std::size_t get_header_size() const { return m_objh->header_size(); }
+
+    coro<std::span<const char>> get(std::span<char> buffer) {
+        std::size_t read_size = 0;
+        address partial_addr;
+        while (m_addr_index < m_objh->get_address().size() &&
+               read_size < buffer.size()) {
+
+            auto frag = m_objh->get_address().get(m_addr_index);
+
+            frag.pointer += m_frag_offset;
+            frag.size -= m_frag_offset;
+
+            if (frag.size + read_size > buffer.size()) {
+                auto remains = buffer.size() - read_size;
+                m_frag_offset += remains;
+                frag.size = remains;
+                partial_addr.push(frag);
+            } else {
+                m_frag_offset = 0;
+                partial_addr.push(frag);
+                m_addr_index++;
+            }
+
+            read_size += frag.size;
+        }
+
+        if (read_size > 0) {
+            co_await m_storage.get().read_address(partial_addr,
+                                                  {buffer.data(), read_size});
+        }
+        co_return std::span<const char>{buffer.data(), read_size};
+    }
+
+private:
+    std::reference_wrapper<storage::data_view> m_storage;
+    std::shared_ptr<object_handle> m_objh;
+
+    std::size_t m_addr_index{0};
+    std::size_t m_frag_offset{0};
+};
+
+} // namespace uh::cluster::proxy::cache::disk
diff --git a/src/proxy/cache/disk/manager.h b/src/proxy/cache/disk/manager.h
index f7eb75ada..cc8f5d688 100644
--- a/src/proxy/cache/disk/manager.h
+++ b/src/proxy/cache/disk/manager.h
@@ -1,7 +1,7 @@
 #pragma once
 
-#include <proxy/cache/disk/body.h>
 #include <proxy/cache/disk/deletion_queue.h>
+#include <proxy/cache/disk/disk_io.h>
 
 #include <proxy/cache/lfu_cache.h>
 #include <proxy/cache/lru_cache.h>
@@ -27,13 +27,8 @@ class manager {
     using stream = ep::http::stream;
     using body = ep::http::body;
 
-    /*
-     * Store object handle in cache
-     *
-     * It removed address information from the given body.
-     */
-    coro<void> put(object_metadata key, reader_body& body) {
-        auto objh = body.get_object_handle();
+    coro<void> put(object_metadata key, disk_sink& w) {
+        auto objh = w.get_object_handle();
         auto obj_size = objh.data_size();
 
         auto total_size =
@@ -66,12 +61,12 @@ class manager {
         std::cout << "Total size after put: " << m_current_size << std::endl;
     }
 
-    std::optional<writer_body> get(object_metadata key) {
+    std::unique_ptr<disk_source> get(object_metadata key) {
         auto entry = m_cache->get(key);
         if (!entry) {
-            return std::nullopt;
+            return nullptr;
         }
-        return writer_body{m_storage, std::move(entry)};
+        return std::make_unique<disk_source>(m_storage, std::move(entry));
     }
 
     static manager create(boost::asio::io_context& ioc, data_view& storage,
@@ -88,7 +83,6 @@ class manager {
     std::atomic<std::size_t> m_current_size{0};
 
     deletion_queue_t m_deletion_queue;
-    // TODO: spawn a background task to remove
     scoped_task m_task;
 
     manager(boost::asio::io_context& ioc, data_view& storage,
diff --git a/src/proxy/cache/disk/object.h b/src/proxy/cache/disk/object.h
index 79e75ab93..f7f054fec 100644
--- a/src/proxy/cache/disk/object.h
+++ b/src/proxy/cache/disk/object.h
@@ -37,18 +37,22 @@ namespace uh::cluster::proxy::cache::disk {
 
 struct object_handle {
     object_handle() = default;
-    object_handle(address&& a)
-        : m_addr(std::move(a)) {}
+    object_handle(address&& a, std::size_t header_size = 0)
+        : m_addr(std::move(a)),
+          m_header_size(header_size) {}
 
     object_handle(object_handle&&) = default;
     object_handle& operator=(object_handle&&) = default;
 
+    std::size_t header_size() const { return m_header_size; }
+
     std::size_t data_size() const { return m_addr.data_size(); }
 
     const address& get_address() const { return m_addr; }
 
 private:
     address m_addr;
+    std::size_t m_header_size;
 };
 
 } // namespace uh::cluster::proxy::cache::disk
diff --git a/src/proxy/handler.cpp b/src/proxy/handler.cpp
index 3534d37c0..f8e1ae5e0 100644
--- a/src/proxy/handler.cpp
+++ b/src/proxy/handler.cpp
@@ -2,22 +2,28 @@
 
 #include "forward_stream.h"
 
+#include <proxy/asio.h>
+#include <proxy/http.h>
+
+#include <proxy/cache/disk/disk_io.h>
+#include <proxy/socket_io.h>
+#include <proxy/tee_io.h>
+
 #include <common/telemetry/metrics.h>
 #include <common/utils/random.h>
 #include <entrypoint/commands/s3/get_object.h>
 #include <entrypoint/http/command_exception.h>
 #include <entrypoint/http/response.h>
 
-using namespace uh::cluster::ep::http;
+using namespace boost::beast;
+using namespace boost::beast::http;
 
 namespace uh::cluster::proxy {
 
 handler::handler(
     std::unique_ptr<request_factory> factory,
     std::function<std::unique_ptr<boost::asio::ip::tcp::socket>()> sf,
-    storage::data_view& dv,
-    cache::disk::manager& mgr,
-    std::size_t buffer_size)
+    storage::data_view& dv, cache::disk::manager& mgr, std::size_t buffer_size)
     : m_factory(std::move(factory)),
       m_sf(std::move(sf)),
       m_dv(dv),
@@ -29,37 +35,43 @@ coro<void> handler::handle(boost::asio::ip::tcp::socket s) {
     auto peer = s.remote_endpoint();
 
     forward_stream incoming(s, *ds);
-    forward_stream outgoing(*ds, s);
-    for (;;) {
+    auto& outgoing{*ds};
+
+    constexpr std::size_t buffer_size_to_load = 16_MiB;
 
+    constexpr std::size_t buffer_size_to_relay_and_store = 32_MiB;
+    constexpr std::size_t buffer_size_to_relay = 4_KiB;
+
+    flat_buffer buffer(
+        std::max(buffer_size_to_relay, buffer_size_to_relay_and_store));
+
+    for (;;) {
         /*
          * Note: lifetime of response must not exceed lifetime of request.
          */
         std::string id = generate_unique_id();
 
-        raw_request rawreq;
-        std::optional<response> resp;
+        ep::http::raw_request rawreq;
+        std::optional<ep::http::response> resp;
 
         try {
-            rawreq = co_await raw_request::read(incoming, peer);
+            rawreq = co_await ep::http::raw_request::read(incoming, peer);
 
             auto& r = rawreq.headers;
             LOG_INFO() << peer << ": incoming request: " << r.method_string()
                        << " " << r.target();
 
             incoming.set_mode(forward_stream::forwarding);
-            outgoing.set_mode(forward_stream::forwarding);
-            std::unique_ptr<request> req =
+            std::unique_ptr<ep::http::request> req =
                 co_await m_factory->create(incoming, rawreq);
 
             if (get_object::can_handle(*req)) {
-                auto writer = m_mgr.get(cache::disk::object_metadata{ req->object_key() });
-                if (writer) {
+                auto d_source =
+                    m_mgr.get(cache::disk::object_metadata{req->object_key()});
+                if (d_source) {
                     LOG_INFO() << peer << ": handling from cache";
                     incoming.set_mode(forward_stream::deleting);
-                    outgoing.set_mode(forward_stream::deleting);
 
-                    // forwarding request
                     auto& b = req->body();
                     auto bs = b.buffer_size();
 
@@ -71,12 +83,19 @@ coro<void> handler::handle(boost::asio::ip::tcp::socket s) {
 
                     LOG_INFO() << peer << ": done reading complete request";
 
-                    std::span<const char> data = co_await writer->get();
-                    while (!data.empty()) {
-                        LOG_INFO() << peer << ": sending " << data.size() << " bytes response";
-                        co_await incoming.write(data);
-                        data = co_await writer->get();
-                    }
+                    response_parser<empty_body> parser;
+                    response_serializer<empty_body, fields> serializer{
+                        parser.get()};
+
+                    co_await async_read_header(d_source, parser);
+
+                    const char* via_value = PROJECT_NAME " " PROJECT_VERSION;
+                    parser.get().set(field::via, via_value);
+
+                    co_await async_write<buffer_size_to_load>(
+                        async_write_header(s, serializer,
+                                           boost::asio::use_awaitable),
+                        s, *d_source);
 
                     LOG_INFO() << peer << ": cache result served";
                     continue;
@@ -90,11 +109,13 @@ coro<void> handler::handle(boost::asio::ip::tcp::socket s) {
                 expect && *expect == "100-continue") {
                 LOG_INFO() << req->peer() << ": forwarding 100 CONTINUE";
                 // TODO timeout
-                co_await outgoing.read_until("\r\n\r\n");
-                co_await outgoing.consume();
+                response_parser<empty_body> p;
+                response_serializer<empty_body, fields> sr{p.get()};
+                co_await async_read_header(outgoing, buffer, p);
+                co_await async_write_header(s, sr);
             }
 
-            // forwarding request
+            // forwarding request body
             auto& b = req->body();
             auto bs = b.buffer_size();
 
@@ -105,59 +126,49 @@ coro<void> handler::handle(boost::asio::ip::tcp::socket s) {
             co_await b.consume();
 
             // forwarding response
-            beast::http::response_parser<beast::http::empty_body> parser;
-            parser.body_limit((std::numeric_limits<std::uint64_t>::max)());
-
-            auto buffer = co_await outgoing.read_until("\r\n\r\n");
-
-            beast::error_code ec;
-            parser.put(boost::asio::buffer(buffer), ec);
-
-            auto res = parser.release();
-
-            bs = outgoing.buffer_size();
-            std::size_t read = 0ull;
-            std::size_t len = std::stoul(res.at("Content-Length"));
-            if (r.method() == boost::beast::http::verb::head &&
-                (res.result_int() / 100 == 2)) {
-                len = 0;
-            }
-
-            LOG_INFO() << peer << ": sending response " << res.result_int()
-                        << " " << res.reason() << " -- " << len;
-
-            if (get_object::can_handle(*req)) {
-                cache::disk::reader_body data(m_dv);
-                LOG_INFO() << peer << ": add " << buffer.size() << " response header";
-                co_await data.put(buffer);
-
-                while (read < len) {
-                    co_await outgoing.consume();
-
-                    auto r = co_await outgoing.read(len - read);
-                    LOG_INFO() << peer << ": add " << r.size() << " response data";
-                    co_await data.put(r);
-
-                    // r: data
-                    read += r.size();
+            response_parser<empty_body> p;
+            p.body_limit(std::numeric_limits<std::uint64_t>::max());
+            response_serializer<empty_body, fields> sr{p.get()};
+
+            LOG_INFO() << peer << ": reading header from downstream";
+            co_await async_read_header(outgoing, buffer, p);
+
+            if (r.method() == verb::head) {
+                LOG_INFO() << peer << ": HEAD request, skipping body relay";
+                co_await async_write_header(s, sr);
+
+            } else if (get_object::can_handle(*req)) {
+                auto d_sink = cache::disk::disk_sink{m_dv};
+                auto s_sink = socket_sink{s};
+                auto body_size = get_content_length(p.get());
+                if (!body_size.has_value()) {
+                    throw std::runtime_error("no content length");
                 }
+                LOG_INFO() << peer << ": relaying and storing body of size "
+                           << *body_size;
+                co_await async_read<buffer_size_to_relay_and_store>(
+                    [&]() -> coro<void> {
+                        auto n = co_await async_write_header(
+                            tee(s_sink, d_sink), sr);
+                        d_sink.set_header_size(n);
+                    },
+                    outgoing, buffer, *body_size, tee(s_sink, d_sink));
+                co_await m_mgr.put(
+                    cache::disk::object_metadata{req->object_key()}, d_sink);
 
-                co_await m_mgr.put(cache::disk::object_metadata{ req->object_key() }, data);
-                co_await outgoing.consume();
             } else {
-                while (read < len) {
-                    co_await outgoing.consume();
-
-                    auto r = co_await outgoing.read(len - read);
-                    // r: data
-                    read += r.size();
+                auto body_size = get_content_length(p.get());
+                if (!body_size.has_value()) {
+                    throw std::runtime_error("no content length");
                 }
-
-                co_await outgoing.consume();
+                LOG_INFO() << peer << ": relaying body of size " << *body_size;
+                co_await async_read<buffer_size_to_relay>(
+                    async_write_header(s, sr, boost::asio::use_awaitable),
+                    outgoing, buffer, *body_size, socket_sink(s));
             }
+            LOG_INFO() << peer << ": done";
 
             metric<success>::increase(1);
-
         } catch (const boost::system::system_error& e) {
             throw;
         } catch (const command_exception& e) {
diff --git a/src/proxy/http.h b/src/proxy/http.h
new file mode 100644
index 000000000..23a9c016b
--- /dev/null
+++ b/src/proxy/http.h
@@ -0,0 +1,263 @@
+#pragma once
+
+#include <common/telemetry/trace/awaitable_operators.h>
+#include <common/types/common_types.h>
+
+#include <proxy/asio.h>
+
+#include <boost/beast/core/flat_buffer.hpp>
+#include <boost/beast/http.hpp>
+
+#include <concepts>
+#include <span>
+
+namespace boost::beast::http {
+
+// The detail namespace means "not public"
+namespace detail {
+
+// This helper is needed for C++11.
+// When invoked with a buffer sequence, writes the buffers `to the
+// std::ostream`.
+template <class Serializer> class write_ostream_helper {
+    Serializer& sr_;
+    std::ostream& os_;
+
+public:
+    write_ostream_helper(Serializer& sr, std::ostream& os)
+        : sr_(sr),
+          os_(os) {}
+
+    // This function is called by the serializer
+    template <class ConstBufferSequence>
+    void operator()(error_code& ec, ConstBufferSequence const& buffers) const {
+        // Error codes must be cleared on success
+        ec = {};
+
+        // Keep a running total of how much we wrote
+        std::size_t bytes_transferred = 0;
+
+        // Loop over the buffer sequence
+        for (auto it = boost::asio::buffer_sequence_begin(buffers);
+             it != boost::asio::buffer_sequence_end(buffers); ++it) {
+            // This is the next buffer in the sequence
+            boost::asio::const_buffer const buffer = *it;
+
+            // Write it to the std::ostream
+            os_.write(reinterpret_cast<char const*>(buffer.data()),
+                      buffer.size());
+
+            // If the std::ostream fails, convert it to an error code
+            if (os_.fail()) {
+                ec = make_error_code(errc::io_error);
+                return;
+            }
+
+            // Adjust our running total
+            bytes_transferred += buffer_size(buffer);
+        }
+
+        // Inform the serializer of the amount we consumed
+        sr_.consume(bytes_transferred);
+    }
+};
+
+} // namespace detail
+
+/** Write a message to a `std::ostream`.
+
+    This function writes the serialized representation of the
+    HTTP/1 message to the sream.
+
+    @param os The `std::ostream` to write to.
+
+    @param msg The message to serialize.
+
+    @param ec Set to the error, if any occurred.
+*/
+template <typename Serializer>
+void write_ostream(std::ostream& os, Serializer& sr, error_code& ec) {
+
+    // This lambda is used as the "visit" function
+    detail::write_ostream_helper<decltype(sr)> lambda{sr, os};
+    do {
+        // In C++14 we could use a generic lambda but since we want
+        // to require only C++11, the lambda is written out by hand.
+        // This function call retrieves the next serialized buffers.
+        sr.next(ec, lambda);
+        if (ec)
+            return;
+    } while (!sr.is_done());
+}
+
+template <typename Message>
+std::optional<std::uint64_t> get_content_length(const Message& msg) {
+    auto it = msg.find(boost::beast::http::field::content_length);
+    if (it != msg.end()) {
+        try {
+            return std::stoull(
+                std::string(it->value().data(), it->value().size()));
+        } catch (...) {
+            return std::nullopt;
+        }
+    }
+    return std::nullopt;
+}
+
+} // namespace boost::beast::http
+
+namespace uh::cluster::proxy {
+
+template <typename SourceType, typename Parser>
+coro<void> async_read_header(const SourceType& source, Parser& parser) {
+    auto header_size = std::vector<char>(source->get_header_size());
+    auto header = co_await source->get(header_size);
+
+    parser.body_limit(std::numeric_limits<std::uint64_t>::max());
+    boost::system::error_code ec;
+    parser.put(boost::asio::const_buffer(header), ec);
+    if (ec) {
+        throw boost::system::system_error(ec);
+    }
+}
+
+template <typename SinkType, typename Serializer>
+coro<std::size_t> async_write_header(SinkType&& sink, Serializer& sr) {
+    using boost::asio::experimental::awaitable_operators::operator&&;
+    std::ostringstream oss;
+    boost::system::error_code ec;
+    sr.split(true);
+    write_ostream(oss, sr, ec);
+    auto header_str = oss.str();
+    if (header_str.size() == 0) {
+        throw std::runtime_error("Could not serialize header");
+    }
+    co_await sink.put(header_str);
+    co_return header_str.size();
+}
+
+template <std::size_t chunk_size, typename Awaitable, typename Incomming,
+          typename SinkType>
+coro<void> async_read(Awaitable&& precursor, Incomming& s,
+                      boost::beast::flat_buffer& b, std::size_t payload_size,
+                      SinkType&& sink) {
+    using boost::asio::experimental::awaitable_operators::operator&&;
+
+    auto sink_ref = std::forward<SinkType>(sink);
+
+    using precursor_type = std::decay_t<Awaitable>;
+    coro<void> precursor_wrapper;
+
+    if constexpr (std::is_same_v<precursor_type, coro<void>>) {
+        precursor_wrapper = std::move(precursor);
+    } else if constexpr (is_boost_awaitable_v<precursor_type>) {
+        precursor_wrapper = async_wrap(std::move(precursor));
+    } else if constexpr (std::is_invocable_r_v<coro<void>, precursor_type>) {
+        precursor_wrapper = precursor();
+    } else {
+        throw std::runtime_error(
+            "invalid precursor type: " +
+            boost::core::demangle(typeid(precursor_type).name()));
+    }
+
+    if (b.data().size() >= payload_size) {
+        co_await std::move(precursor_wrapper);
+        auto sv = std::span<const char>(
+            static_cast<const char*>(b.data().data()), payload_size);
+        co_await sink_ref.put(sv);
+        b.consume(sv.size());
+
+    } else {
+        auto read = [&](auto& s, auto& buffer,
+                        std::size_t required) -> coro<std::size_t> {
+            co_return co_await async_read(s, buffer.prepare(required));
+        };
+        if (payload_size > chunk_size) {
+            boost::beast::flat_buffer b2(chunk_size);
+            auto* rbuf = &b;
+            auto* wbuf = &b2;
+
+            auto remained = payload_size;
+            std::size_t n = 0;
+            if (chunk_size > rbuf->data().size()) {
+                n = co_await (
+                    read(s, *rbuf, chunk_size - rbuf->data().size()) &&
+                    std::move(precursor_wrapper));
+                rbuf->commit(n);
+                remained -= rbuf->data().size();
+            } else {
+                co_await std::move(precursor_wrapper);
+            }
+
+            do {
+                std::swap(rbuf, wbuf);
+                n = co_await (read(s, *rbuf, std::min(remained, chunk_size)) &&
+                              sink_ref.put(get_span(wbuf->data())));
+                rbuf->commit(n);
+                remained -= rbuf->data().size();
+                wbuf->consume(wbuf->data().size());
+            } while (n != 0);
+        } else {
+            auto n = co_await (read(s, b, payload_size - b.data().size()) &&
+                               std::move(precursor_wrapper));
+            b.commit(n);
+            co_await sink_ref.put(get_span(b.data()));
+            b.consume(b.data().size());
+        }
+    }
+
+    b.shrink_to_fit();
+}
+
+template <std::size_t chunk_size, typename Incomming, typename SinkType>
+coro<void> async_read(Incomming& s, boost::beast::flat_buffer& b,
+                      std::size_t payload_size, SinkType&& sink) {
+    co_await async_read<chunk_size>(async_noop(), s, b, payload_size,
+                                    std::forward<SinkType>(sink));
+}
+
+template <std::size_t chunk_size, typename Awaitable, typename SocketType,
+          typename SourceType>
+coro<void> async_write(Awaitable&& precursor, SocketType& s,
+                       SourceType& source) {
+    using boost::asio::experimental::awaitable_operators::operator&&;
+
+    auto source_ref = std::forward<SourceType>(source);
+
+    using precursor_type = std::decay_t<Awaitable>;
+    coro<void> precursor_wrapper;
+
+    if constexpr (std::is_same_v<precursor_type, coro<void>>) {
+        precursor_wrapper = std::move(precursor);
+    } else if constexpr (is_boost_awaitable_v<precursor_type>) {
+        precursor_wrapper = async_wrap(std::move(precursor));
+    } else if constexpr (std::is_invocable_r_v<coro<void>, precursor_type>) {
+        precursor_wrapper = precursor();
+    } else {
+        throw std::runtime_error(
+            "invalid precursor type: " +
+            boost::core::demangle(typeid(precursor_type).name()));
+    }
+
+    char _buf[2][chunk_size];
+    char* rbuf = _buf[0];
+    char* wbuf = _buf[1];
+
+    for (auto data = co_await (source_ref.get({rbuf, chunk_size}) &&
+                               std::move(precursor_wrapper));
+         !data.empty();) {
+        std::swap(rbuf, wbuf);
+        auto d = co_await (source_ref.get({rbuf, chunk_size}) &&
+                               [&]() -> coro<void> {
+            co_await async_write(s, boost::asio::const_buffer(data));
+        }());
+        data = d;
+    }
+}
+
+template <std::size_t chunk_size, typename SocketType, typename SourceType>
+coro<void> async_write(SocketType& s, SourceType& source) {
+    co_await async_write<chunk_size>(async_noop(), s, source);
+}
+
+} // namespace uh::cluster::proxy
diff --git a/src/proxy/socket_io.h b/src/proxy/socket_io.h
new file mode 100644
index 000000000..ec73680f4
--- /dev/null
+++ b/src/proxy/socket_io.h
@@ -0,0 +1,41 @@
+/*
+ * Sync/source, which supports get/put API only
+ */
+#pragma once
+
+#include <common/types/common_types.h>
+
+namespace uh::cluster::proxy {
+
+template <typename SocketType> class socket_sink {
+public:
+    socket_sink(SocketType& s)
+        : m_s{s} {}
+
+    coro<void> put(std::span<const char> sv) {
+        if (sv.size() == 0) {
+            co_return;
+        }
+        co_await boost::asio::async_write(m_s, boost::asio::buffer(sv));
+    }
+
+private:
+    SocketType& m_s;
+};
+
+template <typename SocketType> class socket_source {
+public:
+    socket_source(SocketType& s)
+        : m_s{s} {}
+
+    coro<std::span<const char>> get(std::span<char> buffer) {
+        auto n =
+            co_await boost::asio::async_read(m_s, boost::asio::buffer(buffer));
+        co_return std::span<const char>(buffer.data(), n);
+    }
+
+private:
+    SocketType& m_s;
+};
+
+} // namespace uh::cluster::proxy
diff --git a/src/proxy/tee_io.h b/src/proxy/tee_io.h
new file mode 100644
index 000000000..6cba1c8f2
--- /dev/null
+++ b/src/proxy/tee_io.h
@@ -0,0 +1,28 @@
+#pragma once
+
+#include <common/telemetry/trace/awaitable_operators.h>
+#include <common/types/common_types.h>
+
+namespace uh::cluster::proxy {
+
+template <typename T, typename U> class tee {
+public:
+    tee(T& t, U& u)
+        : m_t{t},
+          m_u{u} {}
+
+    coro<void> put(std::span<const char> sv) {
+        using boost::asio::experimental::awaitable_operators::operator&&;
+
+        if (sv.size() == 0) {
+            co_return;
+        }
+        co_await (m_t.put(sv) && m_u.put(sv));
+    }
+
+private:
+    T& m_t;
+    U& m_u;
+};
+
+} // namespace uh::cluster::proxy
diff --git a/test/unit/test_disk_cache_body.cpp b/test/unit/test_disk_cache_body.cpp
index c860e1764..060584cd3 100644
--- a/test/unit/test_disk_cache_body.cpp
+++ b/test/unit/test_disk_cache_body.cpp
@@ -4,90 +4,22 @@
 
 #include <util/dedupe_fixture.h>
 
-#include <proxy/cache/disk/body.h>
+#include <proxy/cache/disk/disk_io.h>
+#include <proxy/http.h>
+#include <proxy/socket_io.h>
+#include <proxy/tee_io.h>
 
 #include <common/utils/random.h>
-#include <entrypoint/http/stream.h>
-#include <proxy/cache/disk/body.h>
 
 #include <boost/beast/http/parser.hpp>
 #include <boost/beast/http/string_body.hpp>
 
+using namespace boost::beast::http;
+
 namespace uh::cluster::proxy::cache::disk {
 
 BOOST_FIXTURE_TEST_SUITE(a_disk_cache_body, dedupe_fixture)
 
-BOOST_AUTO_TEST_CASE(supports_read) {
-    std::string data = random_string(64);
-    std::string header = "POST /upload HTTP/1.1\r\n"
-                         "Host: localhost\r\n"
-                         "Content-Length: " +
-                         std::to_string(data.size()) + "\r\n\r\n";
-    std::string req = header + data;
-
-    std::cout << req << std::endl;
-
-    // Set up TCP sockets
-    boost::asio::ip::tcp::acceptor acceptor(m_ioc,
-                                            {boost::asio::ip::tcp::v4(), 0});
-    auto endpoint = acceptor.local_endpoint();
-
-    boost::asio::ip::tcp::socket server_sock(m_ioc);
-    boost::asio::ip::tcp::socket client_sock(m_ioc);
-
-    client_sock.connect(endpoint);
-    acceptor.accept(server_sock);
-
-    // Client writes HTTP request
-    auto written_size =
-        boost::asio::write(client_sock, boost::asio::buffer(req));
-
-    BOOST_TEST(written_size == req.size());
-
-    // Read header
-    ep::http::socket_stream stream(server_sock);
-    auto buffer = boost::asio::co_spawn(m_ioc, stream.read_until("\r\n\r\n"),
-                                        boost::asio::use_future)
-                      .get();
-
-    BOOST_TEST(!buffer.empty());
-    BOOST_TEST(buffer.size() == header.size());
-    BOOST_TEST(std::string(buffer.data(), buffer.size()) == header);
-
-    boost::beast::http::request_parser<boost::beast::http::empty_body> parser;
-    parser.body_limit((std::numeric_limits<std::uint64_t>::max)());
-    boost::beast::error_code ec;
-    // parser.put(boost::asio::buffer(buffer), ec);
-    parser.put(boost::asio::buffer(header.data(), header.size()), ec);
-
-    auto res = parser.get();
-
-    BOOST_TEST(parser.is_header_done());
-
-    std::size_t content_length = std::stoul(res.at("Content-Length"));
-
-    BOOST_TEST(content_length == data.size());
-
-    // 6. Read body using async_read and reader_body
-    reader_body body(data_view);
-    boost::asio::co_spawn(m_ioc, async_read(stream, body, content_length),
-                          boost::asio::use_future)
-        .get();
-
-    // 7. Verify body was stored and can be read back
-    auto objh = body.get_object_handle();
-    BOOST_TEST(objh.data_size() == data.size());
-
-    std::vector<char> buf(data.size());
-    boost::asio::co_spawn(
-        m_ioc,
-        data_view.read_address(objh.get_address(),
-                               std::span<char>{buf.data(), buf.size()}),
-        boost::asio::use_future)
-        .get();
-    BOOST_TEST(std::string(buf.data(), buf.size()) == data);
-}
-
 BOOST_AUTO_TEST_CASE(supports_write) {
     std::string data = random_string(64);
     std::string header = "POST /download HTTP/1.1\r\n"
@@ -106,7 +38,7 @@ BOOST_AUTO_TEST_CASE(supports_write) {
     auto objh = std::make_shared<object_handle>(std::move(addr));
 
     BOOST_TEST(objh->data_size() == data.size());
-    writer_body body(data_view, std::move(objh), 16);
+    disk_source source(data_view, std::move(objh));
 
     // Set up TCP sockets
     boost::asio::ip::tcp::acceptor acceptor(m_ioc,
@@ -117,15 +49,13 @@ BOOST_AUTO_TEST_CASE(supports_write) {
     client_sock.connect(endpoint);
     acceptor.accept(server_sock);
 
-    ep::http::socket_stream stream(client_sock);
-
     // Client writes HTTP response header
     auto written_size =
         boost::asio::write(client_sock, boost::asio::buffer(header));
     BOOST_TEST(written_size == header.size());
 
-    // Client writes body using async_write and writer_body
-    boost::asio::co_spawn(m_ioc, async_write(stream, body),
+    // Client writes source using async_write and writer_body
+    boost::asio::co_spawn(m_ioc, async_write<16_KiB>(client_sock, source),
                           boost::asio::use_future)
         .get();
 
@@ -138,8 +68,156 @@ BOOST_AUTO_TEST_CASE(supports_write) {
     BOOST_TEST(received == expected_response);
 }
 
-BOOST_AUTO_TEST_CASE(supports_write_using_smaller_buffer) {}
+BOOST_AUTO_TEST_CASE(goes_with_relay_store_body) {
+    using namespace boost::asio;
+    using namespace boost::beast::http;
+
+    ip::tcp::acceptor acceptor(m_ioc, ip::tcp::endpoint(ip::tcp::v4(), 0));
+    ip::tcp::endpoint endpoint = acceptor.local_endpoint();
+
+    ip::tcp::socket server_socket(m_ioc);
+    ip::tcp::socket client_socket(m_ioc);
+
+    std::thread server_thread([&] { acceptor.accept(server_socket); });
+
+    client_socket.connect(endpoint);
+    server_thread.join();
+
+    std::string body = random_string(8_KiB + 17);
+    std::string header = "POST /upload HTTP/1.1\r\n"
+                         "Host: example.com\r\n"
+                         "User-Agent: test\r\n"
+                         "Content-Length: " +
+                         std::to_string(body.size()) + "\r\n\r\n";
+    auto raw_message = header + body;
+
+    write(client_socket, buffer(header));
+    write(client_socket, buffer(body));
+
+    boost::beast::flat_buffer b;
+
+    parser<true, empty_body> p;
+    serializer<true, empty_body, fields> sr{p.get()};
+
+    disk_sink dsink(data_view);
+    socket_sink ssink(server_socket);
+
+    co_spawn(
+        m_ioc,
+        [&]() -> coro<void> {
+            auto n = co_await async_read_header(server_socket, b, p);
+            auto m = co_await async_write_header(tee(dsink, ssink), sr);
+            dsink.set_header_size(m);
+            BOOST_TEST(n == m);
+            auto body_size = get_content_length(p.get());
+            if (!body_size.has_value()) {
+                throw std::runtime_error("no content length");
+            }
+            co_await async_read<1_KiB>(server_socket, b, *body_size,
+                                       tee(dsink, ssink));
+        },
+        boost::asio::use_future)
+        .get();
+
+    boost::system::error_code ec;
+    std::vector<char> recv_buf(16 * 1024);
+    size_t n = client_socket.read_some(buffer(recv_buf), ec);
+    std::string output_str(recv_buf.data(), n);
+
+    BOOST_CHECK_NE(output_str.find("POST /upload HTTP/1.1"), std::string::npos);
+    BOOST_CHECK_NE(output_str.find("Host: example.com"), std::string::npos);
+    BOOST_CHECK_NE(output_str.find("User-Agent: test"), std::string::npos);
+
+    // auto body_pos = output_str.find("\r\n\r\n");
+    // BOOST_REQUIRE(body_pos != std::string::npos);
+    // body_pos += 4;
+    // std::string_view received_body(&recv_buf[body_pos], n);
+    BOOST_TEST(output_str ==
+               std::string_view(raw_message.data(), raw_message.size()));
+
+    auto objh = dsink.get_object_handle();
+    BOOST_TEST(objh.data_size() == raw_message.size());
+
+    std::vector<char> buf(raw_message.size());
+    boost::asio::co_spawn(
+        m_ioc,
+        data_view.read_address(objh.get_address(),
+                               std::span<char>{buf.data(), buf.size()}),
+        boost::asio::use_future)
+        .get();
+    BOOST_TEST(std::string(buf.data(), buf.size()) == raw_message);
+}
 
 BOOST_AUTO_TEST_SUITE_END()
 
+BOOST_AUTO_TEST_CASE(test_relay_body) {
+    using namespace boost::asio;
+    using namespace boost::beast::http;
+
+    io_context ioc;
+    ip::tcp::acceptor acceptor(ioc, ip::tcp::endpoint(ip::tcp::v4(), 0));
+    ip::tcp::endpoint endpoint = acceptor.local_endpoint();
+
+    ip::tcp::socket server_socket(ioc);
+    ip::tcp::socket client_socket(ioc);
+
+    std::thread server_thread([&] { acceptor.accept(server_socket); });
+
+    client_socket.connect(endpoint);
+    server_thread.join();
+
+    std::string body = random_string(8_KiB + 17);
+    std::string header = "POST /upload HTTP/1.1\r\n"
+                         "Host: example.com\r\n"
+                         "User-Agent: test\r\n"
+                         "Content-Length: " +
+                         std::to_string(body.size()) + "\r\n\r\n";
+
+    write(client_socket, buffer(header));
+    write(client_socket, buffer(body));
+
+    boost::beast::flat_buffer b;
+    auto transform = [](auto&) {};
+
+    auto work_guard = boost::asio::make_work_guard(ioc.get_executor());
+    auto thread = std::thread([&ioc] { ioc.run(); });
+
+    parser<true, empty_body> p;
+    serializer<true, empty_body, fields> sr{p.get()};
+
+    co_spawn(
+        ioc,
+        [&]() -> coro<void> {
+            co_await async_read_header(server_socket, b, p);
+            transform(p.get());
+            co_await async_write_header(server_socket, sr);
+            auto body_size = get_content_length(p.get());
+            if (!body_size.has_value()) {
+                throw std::runtime_error("no content length");
+            }
+            co_await async_read<1_KiB>(server_socket, b, *body_size,
+                                       socket_sink(server_socket));
+        },
+        boost::asio::use_future)
+        .get();
+
+    work_guard.reset();
+    thread.join();
+
+    boost::system::error_code ec;
+    std::vector<char> recv_buf(16 * 1024);
+    size_t n = client_socket.read_some(buffer(recv_buf), ec);
+    std::string output_str(recv_buf.data(), n);
+
+    BOOST_CHECK_NE(output_str.find("POST /upload HTTP/1.1"), std::string::npos);
+    BOOST_CHECK_NE(output_str.find("Host: example.com"), std::string::npos);
+    BOOST_CHECK_NE(output_str.find("User-Agent: test"), std::string::npos);
+
+    auto body_pos = output_str.find("\r\n\r\n");
+    BOOST_REQUIRE(body_pos != std::string::npos);
+    body_pos += 4;
+    std::string_view received_body(&recv_buf[body_pos], n - body_pos);
+    BOOST_TEST(received_body == std::string_view(body.data(), body.size()));
+}
+
 } // namespace uh::cluster::proxy::cache::disk
diff --git a/test/unit/test_disk_cache_manager.cpp b/test/unit/test_disk_cache_manager.cpp
index 660710572..9933a0a94 100644
--- a/test/unit/test_disk_cache_manager.cpp
+++ b/test/unit/test_disk_cache_manager.cpp
@@ -15,10 +15,10 @@ BOOST_AUTO_TEST_CASE(put_and_get_with_metadata) {
     manager mgr{manager::create(m_ioc, data_view, 256)};
 
     std::string data = random_string(64);
-    reader_body rbody(data_view);
+    disk_sink sink(data_view);
 
     boost::asio::co_spawn(
-        m_ioc, rbody.put(std::span<const char>(data.data(), data.size())),
+        m_ioc, sink.put(std::span<const char>(data.data(), data.size())),
         boost::asio::use_future)
         .get();
 
@@ -26,19 +26,19 @@ BOOST_AUTO_TEST_CASE(put_and_get_with_metadata) {
     key.path = "/foo/bar";
     key.version = "v1";
 
-    boost::asio::co_spawn(m_ioc, mgr.put(key, rbody), boost::asio::use_future)
+    boost::asio::co_spawn(m_ioc, mgr.put(key, sink), boost::asio::use_future)
         .get();
 
-    auto wbody_opt = mgr.get(key);
-    BOOST_TEST(wbody_opt.has_value());
+    auto source = mgr.get(key);
+    BOOST_TEST(source != nullptr);
 
-    auto& wbody = wbody_opt.value();
-    auto buf =
-        boost::asio::co_spawn(m_ioc, wbody.get(), boost::asio::use_future)
+    auto buf = std::string(128, '\0');
+    auto sv =
+        boost::asio::co_spawn(m_ioc, source->get(buf), boost::asio::use_future)
             .get();
 
-    BOOST_TEST(buf.size() == data.size());
-    BOOST_TEST(std::string(buf.data(), buf.size()) == data);
+    BOOST_TEST(sv.size() == data.size());
+    BOOST_TEST(std::string(sv.data(), sv.size()) == data);
 }
 
 BOOST_AUTO_TEST_CASE(eviction_test) {
@@ -51,9 +51,9 @@ BOOST_AUTO_TEST_CASE(eviction_test) {
         std::string data = random_string(32);
         datas.push_back(data);
 
-        reader_body rbody(data_view);
+        disk_sink sink(data_view);
         boost::asio::co_spawn(
-            m_ioc, rbody.put(std::span<const char>(data.data(), data.size())),
+            m_ioc, sink.put(std::span<const char>(data.data(), data.size())),
             boost::asio::use_future)
             .get();
 
@@ -62,13 +62,13 @@ BOOST_AUTO_TEST_CASE(eviction_test) {
         key.version = "v" + std::to_string(i);
         keys.push_back(key);
 
-        boost::asio::co_spawn(m_ioc, mgr.put(key, rbody),
+        boost::asio::co_spawn(m_ioc, mgr.put(key, sink),
                               boost::asio::use_future)
             .get();
     }
 
-    auto wbody_opt = mgr.get(keys.front());
-    BOOST_TEST(!wbody_opt.has_value());
+    auto source = mgr.get(keys.front());
+    BOOST_TEST(source == nullptr);
 }
 
 BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/unit/test_storage_group_externals.cpp b/test/unit/test_storage_group_externals.cpp
index c5bbde589..ee02a7dd4 100644
--- a/test/unit/test_storage_group_externals.cpp
+++ b/test/unit/test_storage_group_externals.cpp
@@ -42,7 +42,6 @@ BOOST_AUTO_TEST_CASE(is_watched_well) {
     std::promise<void> p;
     std::future<void> f = p.get_future();
 
-    // TODO: Change lambda input type to void.
     auto subscriber = externals_subscriber(
         m_etcd, group_id, num_storages,
         service_factory<storage_interface>(m_ioc, 2), [&]() { p.set_value(); });