diff --git a/golem-common/src/model/oplog/payload/mod.rs b/golem-common/src/model/oplog/payload/mod.rs index 5ce02c2a5c..9ba9d0c6e0 100644 --- a/golem-common/src/model/oplog/payload/mod.rs +++ b/golem-common/src/model/oplog/payload/mod.rs @@ -715,13 +715,23 @@ impl TryFrom for HostResponseGolemRpcScheduledInvocationCompat { pub trait HostPayloadPair { type Req: Into; - type Resp: Into + TryFrom + Clone + Send + 'static; + type Resp: Into + TryFrom + 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 { diff --git a/golem-worker-executor/src/durable_host/concurrent/call.rs b/golem-worker-executor/src/durable_host/concurrent/call.rs index cad1115ac2..d2e05cd92b 100644 --- a/golem-worker-executor/src/durable_host/concurrent/call.rs +++ b/golem-worker-executor/src/durable_host/concurrent/call.rs @@ -4500,16 +4500,17 @@ async fn prepare_end_entry( )) })?; - // 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( .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}" )) diff --git a/golem-worker-executor/src/durable_host/concurrent/tests.rs b/golem-worker-executor/src/durable_host/concurrent/tests.rs index 6be294b041..f172054855 100644 --- a/golem-worker-executor/src/durable_host/concurrent/tests.rs +++ b/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:::: - 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!(