Avoid cloning durable host call responses

golem-common/src/model/oplog/payload/mod.rs

@@ -715,13 +715,23 @@ impl TryFrom<HostResponse> for HostResponseGolemRpcScheduledInvocationCompat {
 
 pub trait HostPayloadPair {
     type Req: Into<HostRequest>;
-    type Resp: Into<HostResponse> + TryFrom<HostResponse, Error = String> + Clone + Send + 'static;
+    type Resp: Into<HostResponse> + TryFrom<HostResponse, Error = String> + Send + 'static;
 
     const INTERFACE: &'static str;
     const FUNCTION: &'static str;
     const FQFN: &'static str;
 
     const HOST_FUNCTION_NAME: host_functions::HostFunctionName;
+
+    /// Recovers a typed response from the [`HostResponse`] produced from that same response.
+    fn unwrap_own_response(response: HostResponse) -> Self::Resp {
+        Self::Resp::try_from(response).unwrap_or_else(|error| {
+            unreachable!(
+                "host response created for {} could not be converted back: {error}",
+                Self::FQFN
+            )
+        })
+    }
 }
 
 pub mod host_functions {

golem-worker-executor/src/durable_host/concurrent/call.rs

@@ -4500,16 +4500,17 @@ async fn prepare_end_entry<Pair: HostPayloadPair>(
         ))
     })?;
 
-    // Host responses can carry deeply nested schema graphs. Clone and serialize them on a fresh
+    // Host responses can carry deeply nested schema graphs. Serialize them on a fresh
     // blocking-task stack rather than on a Tokio worker stack that may already be deep inside a
     // Wasmtime guest call. Box both the task input and output: passing the response inline in the
-    // blocking task envelope can overflow the caller's stack before the task starts. Keep the typed
-    // cache so this changes neither the oplog representation nor same-process payload-read behavior.
+    // blocking task envelope can overflow the caller's stack before the task starts. Leave the
+    // payload uncached so returning the owned response does not require a deep copy for the oplog.
     let response = Box::new(response);
     let prepared = tokio::task::spawn_blocking(move || {
-        let host_response: HostResponse = response.as_ref().clone().into();
+        let host_response: HostResponse = (*response).into();
         let bytes = golem_common::serialization::serialize(&host_response)?;
-        Ok::<_, String>(Box::new((response, bytes, Arc::new(host_response))))
+        let response = Box::new(Pair::unwrap_own_response(host_response));
+        Ok::<_, String>(Box::new((response, bytes)))
     })
     .await
     .map_err(|err| {
@@ -4518,11 +4519,11 @@ async fn prepare_end_entry<Pair: HostPayloadPair>(
     .map_err(|err| {
         WorkerExecutorError::runtime(format!("failed to serialize durable call response: {err}"))
     })?;
-    let (response, bytes, cached) = *prepared;
+    let (response, bytes) = *prepared;
     let raw_payload = oplog.upload_raw_payload(bytes).await.map_err(|err| {
         WorkerExecutorError::runtime(format!("failed to store durable call response: {err}"))
     })?;
-    let response_payload = raw_payload.into_payload_with_cache(cached).map_err(|err| {
+    let response_payload = raw_payload.into_payload().map_err(|err| {
         WorkerExecutorError::runtime(format!(
             "failed to prepare durable call response payload: {err}"
         ))

golem-worker-executor/src/durable_host/concurrent/tests.rs

@@ -1144,7 +1144,8 @@ async fn access_terminal_end_is_appended_before_cleanup_and_permit_release() {
     // permit must stay held and no cleanup event may become visible; both are released only
     // after the append completes (production releases them via `disarm()` after
     // `end_durable_function_access`, strictly downstream of `wait_terminal()`).
-    use golem_common::model::oplog::HostResponseMonotonicClockTimestamp;
+    use golem_common::model::oplog::HostResponseP3HttpClientConsumeBodyChunk;
+    use golem_common::model::oplog::payload::types::SerializableP3HttpBodyChunk;
 
     let (reached_tx, mut reached_rx) = mpsc::unbounded_channel();
     let gate = Arc::new(tokio::sync::Semaphore::new(0));
@@ -1153,7 +1154,7 @@ async fn access_terminal_end_is_appended_before_cleanup_and_permit_release() {
         .add(OplogEntry::Start {
             timestamp: Timestamp::now_utc(),
             parent_start_index: None,
-            function_name: HostFunctionName::MonotonicClockNow,
+            function_name: HostFunctionName::P3HttpClientConsumeBodyChunk,
             invocation_id: None,
             observational_owner: None,
             request: Some(OplogPayload::Inline(Box::new(HostRequest::NoInput(
@@ -1202,11 +1203,24 @@ async fn access_terminal_end_is_appended_before_cleanup_and_permit_release() {
     .expect("failed to build replay state");
     let completion_marker_recorder =
         CompletionMarkerRecorder::new(persist_oplog.clone(), persist_replay_state);
+    let bytes = vec![42u8; 4096];
+    let original_ptr = bytes.as_ptr() as usize;
     let persist = tokio::spawn(async move {
-        let response = HostResponseMonotonicClockTimestamp { nanos: 42 };
-        let result = DurableCallSession::<host_functions::MonotonicClockNow, NotCancellable>::
-                persist_access_terminal(persist_oplog, completion_marker_recorder, &mut guard, start_idx, response, None)
-            .await;
+        let response = HostResponseP3HttpClientConsumeBodyChunk {
+            chunk: SerializableP3HttpBodyChunk::Data(bytes),
+        };
+        let result = DurableCallSession::<
+            host_functions::P3HttpClientConsumeBodyChunk,
+            NotCancellable,
+        >::persist_access_terminal(
+            persist_oplog,
+            completion_marker_recorder,
+            &mut guard,
+            start_idx,
+            response,
+            None,
+        )
+        .await;
         (result, guard)
     });
 
@@ -1234,17 +1248,37 @@ async fn access_terminal_end_is_appended_before_cleanup_and_permit_release() {
 
     gate.add_permits(1);
     let (result, mut guard) = persist.await.expect("persist task must not panic");
-    result.expect("persisting the terminal must succeed");
+    let response = result.expect("persisting the terminal must succeed");
+    let SerializableP3HttpBodyChunk::Data(bytes) = &response.chunk else {
+        panic!("expected the original data chunk");
+    };
+    assert_eq!(bytes.as_ptr() as usize, original_ptr);
 
     // The terminal is durably appended when the persistence stage returns...
-    {
+    let payload = {
         let entries = oplog.entries.lock().await;
         assert_eq!(entries.len(), 2, "expected [Start, End]");
         match &entries[1] {
-            OplogEntry::End { start_index, .. } => assert_eq!(*start_index, start_idx),
+            OplogEntry::End {
+                start_index,
+                response: Some(payload),
+                ..
+            } => {
+                assert_eq!(*start_index, start_idx);
+                assert!(matches!(
+                    payload,
+                    OplogPayload::SerializedInline { cached: None, .. }
+                ));
+                payload.clone()
+            }
             other => panic!("expected End, got {other:?}"),
         }
-    }
+    };
+    let decoded = oplog
+        .download_payload(payload)
+        .await
+        .expect("uncached response must decode");
+    assert_eq!(decoded, response.into());
     // ...while the guard still owns the permit and nothing has been queued: release happens
     // only at the production `disarm()`, strictly after the terminal.
     assert_eq!(