From 6c9848970e6be0d85fcb407988df54b2d3c40a4a Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 07:07:22 +0000 Subject: [PATCH 1/8] Restore exclusive monotonic clock calls and test real recovery history Amp-Thread-ID: https://ampcode.com/threads/T-01a08615-324b-76fb-883a-9fa3472a5e3a Co-authored-by: Amp --- golem-worker-executor-test-utils/src/lib.rs | 36 +- .../durable_host/clocks/monotonic_clock.rs | 64 ++- golem-worker-executor/src/durable_host/mod.rs | 6 - golem-worker-executor/src/preview2/mod.rs | 1 + golem-worker-executor/src/worker/instance.rs | 68 ++- golem-worker-executor/tests/tool_streaming.rs | 469 ++++++++++-------- .../tool-streaming/rust-caller/src/lib.rs | 74 ++- .../tool-streaming/rust-provider/src/lib.rs | 15 - 8 files changed, 367 insertions(+), 366 deletions(-) diff --git a/golem-worker-executor-test-utils/src/lib.rs b/golem-worker-executor-test-utils/src/lib.rs index 99bc667947..89586b6606 100644 --- a/golem-worker-executor-test-utils/src/lib.rs +++ b/golem-worker-executor-test-utils/src/lib.rs @@ -628,6 +628,19 @@ impl TestWorkerExecutor { ); } + pub async fn commit_oplog(&self, agent_id: &AgentId) -> anyhow::Result<()> { + let owned_agent_id = OwnedAgentId::new(self.context.default_environment_id, agent_id); + let worker = self + .additional_test_deps + .try_get_worker(&owned_agent_id) + .await + .ok_or_else(|| anyhow!("worker is not loaded: {owned_agent_id}"))?; + golem_worker_executor::services::HasOplog::oplog(worker.as_ref()) + .commit(CommitLevel::Always) + .await; + Ok(()) + } + pub async fn commit_oplog_entry_bypassing_worker_status( &self, agent_id: &AgentId, @@ -905,8 +918,8 @@ impl TestWorkerExecutor { .gate_next_entity_body_start(agent_id.clone()) } - /// Pauses the next accessor monotonic-clock `now` call before it starts durability. - pub async fn gate_next_monotonic_clock_now( + /// Commits and pauses the next live monotonic clock call after its real Start, before End. + pub async fn gate_next_monotonic_clock_start( &self, owned_agent_id: &OwnedAgentId, ) -> anyhow::Result { @@ -917,7 +930,7 @@ impl TestWorkerExecutor { .ok_or_else(|| anyhow!("worker {owned_agent_id} is not currently in ActiveAgents"))?; Ok(worker .owner_execution() - .test_gate_next_monotonic_clock_now()) + .test_gate_next_monotonic_clock_start()) } /// Pauses the next exclusive wall-clock `now` call before it starts durability. @@ -933,23 +946,6 @@ impl TestWorkerExecutor { Ok(worker.owner_execution().test_gate_next_wall_clock_now()) } - /// Makes the current generation's next monotonic-clock `now` call return its live value - /// without creating a durable record, so crash-tail tests can commit only earlier work. - pub async fn skip_next_monotonic_clock_now_durability( - &self, - owned_agent_id: &OwnedAgentId, - ) -> anyhow::Result<()> { - let worker = self - .additional_test_deps - .try_get_worker(owned_agent_id) - .await - .ok_or_else(|| anyhow!("worker {owned_agent_id} is not currently in ActiveAgents"))?; - worker - .owner_execution() - .test_skip_next_monotonic_clock_now_durability(); - Ok(()) - } - /// Makes the current generation's next wall-clock `now` call return its live value /// without creating a durable record, so crash-tail tests can commit only earlier work. pub async fn skip_next_wall_clock_now_durability( diff --git a/golem-worker-executor/src/durable_host/clocks/monotonic_clock.rs b/golem-worker-executor/src/durable_host/clocks/monotonic_clock.rs index 2b406f72d6..4fe3dfc653 100644 --- a/golem-worker-executor/src/durable_host/clocks/monotonic_clock.rs +++ b/golem-worker-executor/src/durable_host/clocks/monotonic_clock.rs @@ -30,7 +30,36 @@ use golem_common::model::oplog::{ use wasmtime_wasi::clocks::WasiClocksView as _; use wasmtime_wasi::p2::bindings::clocks::monotonic_clock::Host as WasiMonotonicClockHost; -impl Host for DurableWorkerCtx {} +impl Host for DurableWorkerCtx { + async fn now(&mut self) -> anyhow::Result { + let handle = + DurableCallSession::::start( + self, + HostRequestNoInput {}, + DurableFunctionType::ReadLocal, + ) + .await?; + #[cfg(feature = "test-utils")] + if handle.is_live() + && let Err(interrupt) = self + .owner_execution + .test_after_monotonic_clock_start() + .await + { + let mut handle = handle; + handle.abandon_for_trap(); + return Err(interrupt.into()); + } + let result = handle + .run(self, async |ctx| -> wasmtime::Result<_> { + let mut view = ctx.as_wasi_view(); + let nanos = WasiMonotonicClockHost::now(&mut view.clocks()).await?; + Ok(HostResponseMonotonicClockTimestamp { nanos }) + }) + .await?; + Ok(result.nanos) + } +} fn current_monotonic_time( accessor: &Accessor>>, @@ -51,39 +80,6 @@ fn current_monotonic_resolution( } impl HostWithStore for HasSelf> { - async fn now(accessor: &Accessor) -> anyhow::Result { - #[cfg(feature = "test-utils")] - let (skip_durability, owner_execution) = accessor.with(|mut access| { - let ctx = access.get(); - ( - ctx.test_should_skip_monotonic_clock_now_durability(), - ctx.owner_execution.clone(), - ) - }); - #[cfg(feature = "test-utils")] - if skip_durability { - return Ok(current_monotonic_time(accessor)?); - } - #[cfg(feature = "test-utils")] - owner_execution.test_before_monotonic_clock_now().await; - - let result = - DurableCallSession::::invoke_access( - accessor, - accessor.getter(), - HostRequestNoInput {}, - DurableFunctionType::ReadLocal, - async || { - Ok::<_, anyhow::Error>(HostResponseMonotonicClockTimestamp { - nanos: current_monotonic_time(accessor)?, - }) - }, - ) - .await?; - - Ok(result.nanos) - } - async fn resolution(accessor: &Accessor) -> anyhow::Result { let result = DurableCallSession::< host_functions::MonotonicClockResolution, diff --git a/golem-worker-executor/src/durable_host/mod.rs b/golem-worker-executor/src/durable_host/mod.rs index 69d9253d71..774720364a 100644 --- a/golem-worker-executor/src/durable_host/mod.rs +++ b/golem-worker-executor/src/durable_host/mod.rs @@ -627,12 +627,6 @@ fn validate_unshared_memory_growth( } impl DurableWorkerCtx { - #[cfg(feature = "test-utils")] - pub(crate) fn test_should_skip_monotonic_clock_now_durability(&self) -> bool { - self.owner_execution - .test_should_skip_monotonic_clock_now_durability() - } - #[cfg(feature = "test-utils")] pub(crate) fn test_should_skip_wall_clock_now_durability(&self) -> bool { self.owner_execution diff --git a/golem-worker-executor/src/preview2/mod.rs b/golem-worker-executor/src/preview2/mod.rs index f73fb25747..2068538f6f 100644 --- a/golem-worker-executor/src/preview2/mod.rs +++ b/golem-worker-executor/src/preview2/mod.rs @@ -107,6 +107,7 @@ pub mod p2_monotonic_clock { path: r"../wit", world: "wasi:clocks/imports@0.2.6", imports: { + "wasi:clocks/monotonic-clock.now": async | trappable, "wasi:clocks/monotonic-clock": store | async | trappable, default: async | trappable, }, diff --git a/golem-worker-executor/src/worker/instance.rs b/golem-worker-executor/src/worker/instance.rs index e3f7205b9d..dd3e3688a3 100644 --- a/golem-worker-executor/src/worker/instance.rs +++ b/golem-worker-executor/src/worker/instance.rs @@ -122,12 +122,10 @@ pub struct OwnerExecution { deferred_tool_admission: Arc, reached_oplog_marker: AtomicU64, #[cfg(feature = "test-utils")] - monotonic_clock_now_gate: Mutex>>, + monotonic_clock_start_gate: Mutex>>, #[cfg(feature = "test-utils")] wall_clock_now_gate: Mutex>>, #[cfg(feature = "test-utils")] - skip_monotonic_clock_now_durability: AtomicBool, - #[cfg(feature = "test-utils")] skip_wall_clock_now_durability: AtomicBool, } @@ -135,6 +133,7 @@ pub struct OwnerExecution { struct ClockNowGate { entered: Mutex>>, release: tokio::sync::Semaphore, + abort_as_restart: AtomicBool, } #[cfg(feature = "test-utils")] @@ -154,6 +153,11 @@ impl ClockNowGateHandle { pub fn release(&self) { self.gate.release.add_permits(1); } + + pub fn abort_as_restart(&self) { + self.gate.abort_as_restart.store(true, Ordering::Release); + self.release(); + } } #[cfg(feature = "test-utils")] @@ -182,12 +186,10 @@ impl OwnerExecution { deferred_tool_admission: Arc::new(DeferredAdmissionTable::default()), reached_oplog_marker: AtomicU64::new(OplogIndex::NONE.into()), #[cfg(feature = "test-utils")] - monotonic_clock_now_gate: Mutex::new(None), + monotonic_clock_start_gate: Mutex::new(None), #[cfg(feature = "test-utils")] wall_clock_now_gate: Mutex::new(None), #[cfg(feature = "test-utils")] - skip_monotonic_clock_now_durability: AtomicBool::new(false), - #[cfg(feature = "test-utils")] skip_wall_clock_now_durability: AtomicBool::new(false), } } @@ -337,16 +339,37 @@ impl OwnerExecution { #[cfg(feature = "test-utils")] #[doc(hidden)] - pub fn test_gate_next_monotonic_clock_now(&self) -> ClockNowGateHandle { + pub fn test_gate_next_monotonic_clock_start(&self) -> ClockNowGateHandle { let (entered_tx, entered) = tokio::sync::oneshot::channel(); let gate = Arc::new(ClockNowGate { entered: Mutex::new(Some(entered_tx)), release: tokio::sync::Semaphore::new(0), + abort_as_restart: AtomicBool::new(false), }); - *self.monotonic_clock_now_gate.lock().unwrap() = Some(gate.clone()); + *self.monotonic_clock_start_gate.lock().unwrap() = Some(gate.clone()); ClockNowGateHandle { entered, gate } } + #[cfg(feature = "test-utils")] + pub(crate) async fn test_after_monotonic_clock_start(&self) -> Result<(), InterruptKind> { + let gate = self.monotonic_clock_start_gate.lock().unwrap().take(); + if let Some(gate) = gate { + self.oplog.commit(CommitLevel::Always).await; + if let Some(entered) = gate.entered.lock().unwrap().take() { + let _ = entered.send(()); + } + gate.release + .acquire() + .await + .expect("clock Start gate closed") + .forget(); + if gate.abort_as_restart.load(Ordering::Acquire) { + return Err(InterruptKind::Restart); + } + } + Ok(()) + } + #[cfg(feature = "test-utils")] #[doc(hidden)] pub fn test_gate_next_wall_clock_now(&self) -> ClockNowGateHandle { @@ -354,18 +377,12 @@ impl OwnerExecution { let gate = Arc::new(ClockNowGate { entered: Mutex::new(Some(entered_tx)), release: tokio::sync::Semaphore::new(0), + abort_as_restart: AtomicBool::new(false), }); *self.wall_clock_now_gate.lock().unwrap() = Some(gate.clone()); ClockNowGateHandle { entered, gate } } - #[cfg(feature = "test-utils")] - #[doc(hidden)] - pub fn test_skip_next_monotonic_clock_now_durability(&self) { - self.skip_monotonic_clock_now_durability - .store(true, Ordering::Release); - } - #[cfg(feature = "test-utils")] #[doc(hidden)] pub fn test_skip_next_wall_clock_now_durability(&self) { @@ -373,33 +390,12 @@ impl OwnerExecution { .store(true, Ordering::Release); } - #[cfg(feature = "test-utils")] - pub(crate) fn test_should_skip_monotonic_clock_now_durability(&self) -> bool { - self.skip_monotonic_clock_now_durability - .swap(false, Ordering::AcqRel) - } - #[cfg(feature = "test-utils")] pub(crate) fn test_should_skip_wall_clock_now_durability(&self) -> bool { self.skip_wall_clock_now_durability .swap(false, Ordering::AcqRel) } - #[cfg(feature = "test-utils")] - pub(crate) async fn test_before_monotonic_clock_now(&self) { - let gate = self.monotonic_clock_now_gate.lock().unwrap().take(); - if let Some(gate) = gate { - if let Some(entered) = gate.entered.lock().unwrap().take() { - let _ = entered.send(()); - } - gate.release - .acquire() - .await - .expect("monotonic-clock now gate was closed") - .forget(); - } - } - #[cfg(feature = "test-utils")] pub(crate) async fn test_before_wall_clock_now(&self) { let gate = self.wall_clock_now_gate.lock().unwrap().take(); diff --git a/golem-worker-executor/tests/tool_streaming.rs b/golem-worker-executor/tests/tool_streaming.rs index 122a6fed58..a2a3781882 100644 --- a/golem-worker-executor/tests/tool_streaming.rs +++ b/golem-worker-executor/tests/tool_streaming.rs @@ -13,7 +13,6 @@ // limitations under the License. use crate::Tracing; -use anyhow::Context; use axum::Router; use axum::body::{Body, Bytes}; use axum::extract::{Path, Request, State}; @@ -45,9 +44,8 @@ use golem_common::{ }; use golem_test_framework::dsl::TestDsl; use golem_worker_executor::durable_host::tool::{ - ToolAttachmentModeMetadata, ToolAttachmentTerminalMetadata, ToolBodyAdmissionMetadata, - ToolOperationLaneMetadata, ToolOperationMetadata, ToolOperationWinnerMetadata, - ToolOwnerFailureMetadata, + ToolAttachmentModeMetadata, ToolBodyAdmissionMetadata, ToolOperationLaneMetadata, + ToolOperationMetadata, ToolOperationWinnerMetadata, ToolOwnerFailureMetadata, }; use golem_worker_executor::services::environment_state::{ EnvironmentStateService, ToolActivationOutcome, ToolDiscoveryError, @@ -106,6 +104,13 @@ struct StreamEvidence { completion: String, } +#[derive(Debug, FromSchema)] +struct ClockedStreamEvidence { + before_tool_nanos: u64, + after_tool_nanos: u64, + stream: StreamEvidence, +} + #[derive(Debug, FromSchema)] struct TsStreamEvidence { output: Vec, @@ -553,52 +558,6 @@ async fn wait_for_active_tool_operations( Ok(()) } -async fn wait_for_tool_stdin_state( - executor: &TestWorkerExecutor, - agent_id: &OwnedAgentId, - accepted_bytes: u64, - delivered_bytes: u64, - buffered_bytes: usize, - capacity_bytes: usize, - backpressured: bool, - terminal: Option, - producer_operation_active: bool, - producer_active: bool, - consumer_active: bool, -) -> anyhow::Result<()> { - let result = tokio::time::timeout(std::time::Duration::from_secs(30), async { - loop { - if let Some(active) = executor.active_entity_metadata(agent_id).await - && let Some(stdin) = active - .tool_operations - .operations - .first() - .and_then(|operation| operation.stdin.as_ref()) - && stdin.accepted_bytes == accepted_bytes - && stdin.delivered_bytes == delivered_bytes - && stdin.buffered_bytes == buffered_bytes - && stdin.capacity_bytes == capacity_bytes - && stdin.backpressured == backpressured - && stdin.terminal == terminal - && stdin.producer_operation_active == producer_operation_active - && stdin.producer_active == producer_active - && stdin.consumer_active == consumer_active - { - return; - } - tokio::task::yield_now().await; - } - }) - .await; - if result.is_err() { - let active = executor.active_entity_metadata(agent_id).await; - anyhow::bail!( - "timed out waiting for tool stdin state accepted={accepted_bytes}, delivered={delivered_bytes}, buffered={buffered_bytes}, capacity={capacity_bytes}, backpressured={backpressured}, terminal={terminal:?}, producer-operation-active={producer_operation_active}, producer-active={producer_active}, consumer-active={consumer_active}; active metadata: {active:#?}" - ); - } - Ok(()) -} - async fn wait_for_owner_replay_settling( executor: &TestWorkerExecutor, agent_id: &OwnedAgentId, @@ -3293,7 +3252,6 @@ async fn active_stream_crash_replays_pinned_activation_with_fresh_attachments( enum CompletedReconstructionExclusiveCase { Success, Divergence, - BackpressuredStdin, } async fn run_completed_reconstruction_exclusive_p2_case( @@ -3347,9 +3305,6 @@ async fn run_completed_reconstruction_exclusive_p2_case( let case_name = match case { CompletedReconstructionExclusiveCase::Success => "exclusive-p2-success", CompletedReconstructionExclusiveCase::Divergence => "exclusive-p2-divergence", - CompletedReconstructionExclusiveCase::BackpressuredStdin => { - "exclusive-p2-backpressured-stdin" - } }; let agent_id = agent_id!("ToolStreamingCaller", case_name); let worker_id = executor @@ -3370,69 +3325,17 @@ async fn run_completed_reconstruction_exclusive_p2_case( .await .map_err(|_| anyhow::anyhow!("timed out waiting for caller initialization"))??; let owned_agent_id = OwnedAgentId::new(context.default_environment_id, &worker_id); - let first = vec![0x31u8; 64]; - let second = vec![0x32u8; 64]; - let mut original_start = (case == CompletedReconstructionExclusiveCase::BackpressuredStdin) - .then(|| executor.gate_next_entity_body_start(&worker_id)); let mut original_success = executor.gate_next_agent_invocation_success(&worker_id); - match case { - CompletedReconstructionExclusiveCase::Success - | CompletedReconstructionExclusiveCase::Divergence => { - executor - .skip_next_wall_clock_now_durability(&owned_agent_id) - .await?; - } - CompletedReconstructionExclusiveCase::BackpressuredStdin => { - executor - .skip_next_monotonic_clock_now_durability(&owned_agent_id) - .await?; - } - } - let invocation = match case { - CompletedReconstructionExclusiveCase::Success - | CompletedReconstructionExclusiveCase::Divergence => executor.invoke_and_await_agent( - &caller_component, - &agent_id, - "hold_completed_reconstruction_before_exclusive_clock", - data_value!(), - ), - CompletedReconstructionExclusiveCase::BackpressuredStdin => executor - .invoke_and_await_agent( - &caller_component, - &agent_id, - "hold_reconstruction_backpressure_before_exclusive_clock", - data_value!(first.clone(), second.clone()), - ), - }; - tokio::pin!(invocation); - - if let Some(start) = original_start.as_mut() { - tokio::select! { - () = start.entered() => {} - result = &mut invocation => { - result.context("original invocation finished before reaching the entity body start gate")?; - anyhow::bail!("original invocation succeeded without reaching the entity body start gate"); - } - () = tokio::time::sleep(std::time::Duration::from_secs(30)) => { - anyhow::bail!("original entity body start gate was not reached"); - } - } - wait_for_tool_stdin_state( - &executor, - &owned_agent_id, - first.len() as u64, - 0, - first.len(), - first.len(), - true, - None, - true, - true, - true, - ) + executor + .skip_next_wall_clock_now_durability(&owned_agent_id) .await?; - start.release(); - } + let invocation = executor.invoke_and_await_agent( + &caller_component, + &agent_id, + "hold_completed_reconstruction_before_exclusive_clock", + data_value!(), + ); + tokio::pin!(invocation); let validate_recovery = async { tokio::time::timeout( @@ -3464,11 +3367,8 @@ async fn run_completed_reconstruction_exclusive_p2_case( if case == CompletedReconstructionExclusiveCase::Divergence { executor.diverge_next_completed_entity_reconstruction(&worker_id); } - let mut replayed_start = (case == CompletedReconstructionExclusiveCase::BackpressuredStdin) - .then(|| executor.gate_next_entity_body_start(&worker_id)); let mut replayed_claim = executor.gate_next_entity_reconstruction_claim(&worker_id); executor.simulated_crash(&worker_id).await?; - drop(original_start); original_success.abort_as_restart(); drop(original_success); let claimed_start = @@ -3476,17 +3376,7 @@ async fn run_completed_reconstruction_exclusive_p2_case( .await .map_err(|_| anyhow::anyhow!("replayed reconstruction claim was not reached"))?; assert_eq!(claimed_start, reconstruction_start); - let mut replayed_clock = match case { - CompletedReconstructionExclusiveCase::Success - | CompletedReconstructionExclusiveCase::Divergence => { - executor.gate_next_wall_clock_now(&owned_agent_id).await? - } - CompletedReconstructionExclusiveCase::BackpressuredStdin => { - executor - .gate_next_monotonic_clock_now(&owned_agent_id) - .await? - } - }; + let mut replayed_clock = executor.gate_next_wall_clock_now(&owned_agent_id).await?; replayed_claim.release(); tokio::time::timeout(std::time::Duration::from_secs(30), replayed_clock.entered()) .await @@ -3571,72 +3461,6 @@ async fn run_completed_reconstruction_exclusive_p2_case( "divergent reconstruction permitted ReplayFinished update finalization" ); } - CompletedReconstructionExclusiveCase::BackpressuredStdin => { - let replayed_success = replayed_success.as_mut().unwrap(); - let replayed_start = replayed_start.as_mut().unwrap(); - tokio::time::timeout(std::time::Duration::from_secs(30), replayed_start.entered()) - .await - .map_err(|_| { - anyhow::anyhow!("replayed entity body start gate was not reached") - })?; - wait_for_tool_stdin_state( - &executor, - &owned_agent_id, - first.len() as u64, - 0, - first.len(), - first.len(), - true, - None, - true, - true, - true, - ) - .await?; - replayed_start.release(); - wait_for_tool_stdin_state( - &executor, - &owned_agent_id, - (first.len() + second.len()) as u64, - (first.len() + second.len()) as u64, - 0, - first.len(), - false, - Some(ToolAttachmentTerminalMetadata::ConsumerCancelled), - false, - true, - false, - ) - .await?; - tokio::time::timeout( - std::time::Duration::from_secs(30), - reconstruction_body.entered(), - ) - .await - .map_err(|_| { - anyhow::anyhow!("backpressured reconstruction did not reach body validation") - })?; - replayed_clock.release(); - wait_for_owner_replay_settling(&executor, &owned_agent_id).await?; - assert!(!executor.owner_replay_is_live(&owned_agent_id).await?); - assert!( - tokio::time::timeout( - std::time::Duration::from_millis(250), - replayed_success.entered() - ) - .await - .is_err(), - "Store pumping published live before completed body validation" - ); - reconstruction_body.release(); - tokio::time::timeout( - std::time::Duration::from_secs(30), - replayed_success.entered(), - ) - .await - .map_err(|_| anyhow::anyhow!("replayed agent invocation did not finish"))?; - replayed_success.release(); - } } Ok::<_, anyhow::Error>(()) }; @@ -3699,26 +3523,6 @@ async fn completed_reconstruction_divergence_fails_exclusive_p2_wait( .await } -#[test] -#[tracing::instrument] -#[timeout("5m")] -async fn settling_accessor_p2_keeps_backpressured_reconstruction_store_polling( - last_unique_id: &LastUniqueId, - deps: &WorkerExecutorTestDependencies, - #[tagged_as("tool_streaming_rust_provider")] provider: &PrecompiledComponent, - #[tagged_as("tool_streaming_rust_caller")] caller: &PrecompiledComponent, - _tracing: &Tracing, -) -> anyhow::Result<()> { - run_completed_reconstruction_exclusive_p2_case( - last_unique_id, - deps, - provider, - caller, - CompletedReconstructionExclusiveCase::BackpressuredStdin, - ) - .await -} - #[test] #[tracing::instrument] #[timeout("5m")] @@ -4091,6 +3895,241 @@ async fn incomplete_custom_durability_waits_for_completed_reconstruction( Ok(()) } +#[test] +#[tracing::instrument] +#[timeout("5m")] +async fn recorded_monotonic_clock_replays_across_incomplete_stream_recovery( + last_unique_id: &LastUniqueId, + deps: &WorkerExecutorTestDependencies, + #[tagged_as("tool_streaming_rust_provider")] provider: &PrecompiledComponent, + #[tagged_as("tool_streaming_rust_caller")] caller: &PrecompiledComponent, + _tracing: &Tracing, +) -> anyhow::Result<()> { + run_clocked_stream_recovery(last_unique_id, deps, provider, caller, false).await +} + +#[test] +#[tracing::instrument] +#[timeout("2m")] +async fn incomplete_monotonic_clock_recovers_after_completed_stream( + last_unique_id: &LastUniqueId, + deps: &WorkerExecutorTestDependencies, + #[tagged_as("tool_streaming_rust_provider")] provider: &PrecompiledComponent, + #[tagged_as("tool_streaming_rust_caller")] caller: &PrecompiledComponent, + _tracing: &Tracing, +) -> anyhow::Result<()> { + run_clocked_stream_recovery(last_unique_id, deps, provider, caller, true).await +} + +async fn run_clocked_stream_recovery( + last_unique_id: &LastUniqueId, + deps: &WorkerExecutorTestDependencies, + provider: &PrecompiledComponent, + caller: &PrecompiledComponent, + incomplete_clock: bool, +) -> anyhow::Result<()> { + let context = TestContext::new(last_unique_id); + let environment_state = Arc::new(TestEnvironmentStateService::default()); + let executor = start_with_overrides( + deps, + &context, + TestExecutorOverrides { + environment_state_service: Some(environment_state.clone()), + ..Default::default() + }, + ) + .await?; + let (checkpoint_port, checkpoint_gate_port, checkpoint_server, mut checkpoint_arrivals) = + start_crash_checkpoint_server().await; + let provider_component = executor + .component_dep(&context.default_environment_id, provider) + .store() + .await?; + let caller_component = executor + .component_dep(&context.default_environment_id, caller) + .store() + .await?; + let provider_path = deps + .component_directory + .join(format!("{}.wasm", provider.wasm_name)); + let metadata = extract_component_metadata(&provider_path, false, true).await?; + environment_state.set_tool_deployment( + context.default_environment_id, + caller_component.id, + caller_component.revision, + Some(deployment_state( + context.account_id, + provider_component.id, + provider_component.revision, + "golem-it:tool-streaming-rust-provider", + "ToolStreamingCaller", + metadata.tools, + )), + ); + + let agent_id = agent_id!("ToolStreamingCaller", "clocked-incomplete-stream"); + let worker_id = executor + .start_agent_with( + &caller_component.id, + agent_id.clone(), + HashMap::from([ + ( + "CRASH_CHECKPOINT_PORT".to_string(), + checkpoint_port.to_string(), + ), + ( + "CRASH_CHECKPOINT_GATE_PORT".to_string(), + checkpoint_gate_port.to_string(), + ), + ]), + Vec::new(), + ) + .await?; + let first: Vec = (0..64).collect(); + let second: Vec = (192..255).collect(); + let expected = [first.as_slice(), second.as_slice()].concat(); + let mut original_body_start = + (!incomplete_clock).then(|| executor.gate_next_entity_body_start(&worker_id)); + let path = if incomplete_clock { + "hold-body:/clocked-stream.bin" + } else { + "/clocked-stream.bin" + }; + let invocation = executor.invoke_and_await_agent( + &caller_component, + &agent_id, + "clocked_capable_checkpoint", + data_value!(path, first, second), + ); + tokio::pin!(invocation); + + let original_checkpoint = tokio::select! { + checkpoint = async { + if let Some(gate) = original_body_start.as_mut() { + gate.entered().await; + Ok(None) + } else { + checkpoint_arrivals.recv().await.map(Some) + .ok_or_else(|| anyhow::anyhow!("crash checkpoint server stopped")) + } + } => checkpoint?, + result = &mut invocation => { + anyhow::bail!("clocked invocation finished before checkpoint: {result:?}"); + } + () = tokio::time::sleep(std::time::Duration::from_secs(30)) => { + anyhow::bail!("clocked tool did not reach its original body checkpoint"); + } + }; + if let Some(checkpoint) = &original_checkpoint { + assert_eq!(checkpoint.name, "capable-body"); + } + executor.commit_oplog(&worker_id).await?; + let original_oplog = executor.get_oplog(&worker_id, OplogIndex::INITIAL).await?; + let clock_starts: Vec<_> = original_oplog + .iter() + .filter_map(|entry| match &entry.entry { + PublicOplogEntry::Start(params) if params.function_name == "monotonic_clock::now" => { + Some(entry.oplog_index) + } + _ => None, + }) + .collect(); + #[derive(FromSchema)] + struct ClockTimestamp { + nanos: u64, + } + let clocks: Vec = original_oplog + .iter() + .filter_map(|entry| match &entry.entry { + PublicOplogEntry::End(params) if clock_starts.contains(¶ms.start_index) => Some( + ClockTimestamp::from_value( + params + .response + .as_ref() + .expect("clock End response") + .value(), + ) + .map(|time| time.nanos), + ), + _ => None, + }) + .collect::>()?; + assert!( + clocks.len() >= 2, + "normal clock calls must be recorded before the tool" + ); + let recorded_elapsed = clocks[clocks.len() - 1].saturating_sub(clocks[clocks.len() - 2]); + let incomplete_start = if incomplete_clock { + let owner = OwnedAgentId::new(context.default_environment_id, &worker_id); + let mut clock = executor.gate_next_monotonic_clock_start(&owner).await?; + original_checkpoint + .expect("capable body checkpoint") + .release + .send(()) + .expect("release original tool body"); + tokio::time::timeout(std::time::Duration::from_secs(30), clock.entered()) + .await + .map_err(|_| { + anyhow::anyhow!("clock did not persist its Start after the completed tool") + })?; + let oplog = executor.get_oplog(&worker_id, OplogIndex::INITIAL).await?; + let start = oplog + .iter() + .rev() + .find_map(|entry| match &entry.entry { + PublicOplogEntry::Start(params) + if params.function_name == "monotonic_clock::now" => + { + Some(entry.oplog_index) + } + _ => None, + }) + .expect("persisted clock Start"); + assert!(oplog.iter().all(|entry| !matches!(&entry.entry, PublicOplogEntry::End(params) if params.start_index == start))); + wait_for_completed_entity_terminal(&executor, &worker_id).await?; + clock.abort_as_restart(); + Some(start) + } else { + executor.simulated_crash(&worker_id).await?; + drop(original_body_start); + None + }; + + let evidence: ClockedStreamEvidence = + tokio::time::timeout(std::time::Duration::from_secs(30), &mut invocation) + .await + .map_err(|_| { + anyhow::anyhow!("clocked stream recovery deadlocked after checkpoint release") + })?? + .into_typed()?; + assert_eq!(evidence.before_tool_nanos, recorded_elapsed); + assert!(evidence.after_tool_nanos >= evidence.before_tool_nanos); + let expected_output = if incomplete_clock { + [b"body-checkpoint".as_slice(), expected.as_slice()].concat() + } else { + expected.clone() + }; + assert_evidence(&evidence.stream, &expected_output, 2, expected.len() as u64); + assert_eq!( + executor + .get_file_contents(&worker_id, "/clocked-stream.bin") + .await?, + expected + ); + let oplog = executor.get_oplog(&worker_id, OplogIndex::INITIAL).await?; + if let Some(start) = incomplete_start { + assert_eq!(oplog.iter().filter(|entry| matches!(&entry.entry, PublicOplogEntry::End(params) if params.start_index == start)).count(), 1, "incomplete clock must be repaired exactly once"); + } + assert!( + oplog + .iter() + .all(|entry| !matches!(entry.entry, PublicOplogEntry::Error(_))), + "clock and stream recovery must not write a replay error" + ); + checkpoint_server.abort(); + Ok(()) +} + #[test] #[tracing::instrument] #[timeout("8m")] diff --git a/test-components/tool-streaming/rust-caller/src/lib.rs b/test-components/tool-streaming/rust-caller/src/lib.rs index 23fc803522..5c08dac7c1 100644 --- a/test-components/tool-streaming/rust-caller/src/lib.rs +++ b/test-components/tool-streaming/rust-caller/src/lib.rs @@ -6,7 +6,7 @@ use golem_rust::agentic::{ }; use golem_rust::durability::{Durability, DurableFunctionType}; use golem_rust::golem_agentic::golem::tool::host::{ - self as tool_host, ByteStreamCloseCause, ByteStreamFailure, RpcError, ToolRpc, + self as tool_host, ByteStreamFailure, RpcError, ToolRpc, }; use golem_rust::{ FromSchema, IntoSchema, IntoTypedSchemaValue, agent_definition, agent_implementation, @@ -30,6 +30,13 @@ pub struct StreamingBenchmarkResult { pub chunks_read: u32, } +#[derive(Debug, Clone, IntoSchema, FromSchema)] +pub struct ClockedStreamEvidence { + pub before_tool_nanos: u64, + pub after_tool_nanos: u64, + pub stream: StreamEvidence, +} + #[derive(IntoSchema)] struct RawRunInput { mode: String, @@ -106,11 +113,12 @@ pub trait ToolStreamingCaller { ); async fn reject_incomplete_attachment_upgrade_under_pressure(&self) -> Vec; async fn hold_completed_reconstruction_before_exclusive_clock(&self); - async fn hold_reconstruction_backpressure_before_exclusive_clock( + async fn clocked_capable_checkpoint( &self, + path: String, first: Vec, second: Vec, - ); + ) -> ClockedStreamEvidence; async fn hold_completed_reconstruction_before_incomplete_custom(&self); async fn principal_context(&self, principal: Principal) -> Vec; } @@ -1644,47 +1652,33 @@ impl ToolStreamingCaller for ToolStreamingCallerImpl { (tool, exclusive_clock).join().await; } - async fn hold_reconstruction_backpressure_before_exclusive_clock( + async fn clocked_capable_checkpoint( &self, + path: String, first: Vec, second: Vec, - ) { - let rpc = ToolRpc::new("streaming"); - let (stdin_writer, stdin, stdin_closed) = tool_host::create_stdin(); - stdin_writer - .write(first) + ) -> ClockedStreamEvidence { + let started = std::time::Instant::now(); + let before_tool_nanos = started.elapsed().as_nanos() as u64; + let invocation = CapableStreamingClient::default() + .run_capable(path, input_stream(vec![first, second])) + .expect("start clocked capable streaming tool"); + let (summary, output) = invocation + .collect() .await - .expect("prefill backpressured reconstruction stdin"); - let (stdout_target, stdout) = tool_host::create_stdout(); - let result = rpc.async_invoke_and_await( - &["run".to_string()], - raw_input("historical-reconstruction-backpressure"), - Some(stdin), - Some(stdout_target), - ); - let stdin = async { - stdin_writer - .write(second) - .await - .expect("write second backpressured reconstruction stdin chunk"); - }; - let stdin_terminal = async { - assert!(matches!( - stdin_closed.wait().await, - ByteStreamCloseCause::ConsumerCancelled - )); - }; - let tool = async { - assert!(read_all(stdout).await.is_empty()); - raw_result(&result) - .await - .expect("backpressured reconstruction result before exclusive clock call"); - }; - let exclusive_clock = async { - let _ = std::time::Instant::now(); - }; - (stdin, stdin_terminal, tool, exclusive_clock).join().await; - drop(stdin_writer); + .expect("complete clocked capable streaming tool"); + let after_tool_nanos = started.elapsed().as_nanos() as u64; + ClockedStreamEvidence { + before_tool_nanos, + after_tool_nanos, + stream: StreamEvidence { + output, + chunks_read: summary.chunks_read, + bytes_read: summary.bytes_read, + output_closed: summary.output_closed, + completion: "ok".to_string(), + }, + } } async fn hold_completed_reconstruction_before_incomplete_custom(&self) { diff --git a/test-components/tool-streaming/rust-provider/src/lib.rs b/test-components/tool-streaming/rust-provider/src/lib.rs index 5bf662bf66..67e7aaf513 100644 --- a/test-components/tool-streaming/rust-provider/src/lib.rs +++ b/test-components/tool-streaming/rust-provider/src/lib.rs @@ -661,21 +661,6 @@ impl Streaming for StreamingImpl { let _ = stdout.finish().await; return Ok(summary); } - "historical-reconstruction-backpressure" => { - for expected in [vec![0x31; 64], vec![0x32; 64]] { - let chunk = stdin - .next() - .await - .expect("backpressured reconstruction stdin ended early") - .expect("backpressured reconstruction stdin failed"); - assert_eq!(chunk, expected); - summary.chunks_read += 1; - summary.bytes_read += chunk.len() as u64; - } - drop(stdin); - let _ = stdout.finish().await; - return Ok(summary); - } _ => { while let Some(item) = stdin.next().await { let Ok(chunk) = item else { From a5c5968e6ef78800536563153238904ea4dec597 Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 08:47:06 +0000 Subject: [PATCH 2/8] Reproduce snapshot recovery with a core initializer clock read Amp-Thread-ID: https://ampcode.com/threads/T-01a08a68-f1d0-7572-8122-6ea73577b108 Co-authored-by: Amp --- golem-worker-executor/tests/durability.rs | 10 ++++++++++ test-components/agent-counters/src/snapshot_test.rs | 6 ++++++ 2 files changed, 16 insertions(+) diff --git a/golem-worker-executor/tests/durability.rs b/golem-worker-executor/tests/durability.rs index 44ce8406f5..cbd8593794 100644 --- a/golem-worker-executor/tests/durability.rs +++ b/golem-worker-executor/tests/durability.rs @@ -711,6 +711,16 @@ async fn automatic_snapshot_every_2nd_invocation( } let oplog = executor.get_oplog(&worker_id, OplogIndex::INITIAL).await?; + assert!( + oplog + .iter() + .take_while(|entry| !matches!(&entry.entry, PublicOplogEntry::Snapshot(_))) + .any(|entry| matches!( + &entry.entry, + PublicOplogEntry::Start(params) if params.function_name == "monotonic_clock::now" + )), + "core initializer must record a clock call before the snapshot" + ); let snapshot_count = oplog .iter() .filter(|entry| matches!(&entry.entry, PublicOplogEntry::Snapshot(_))) diff --git a/test-components/agent-counters/src/snapshot_test.rs b/test-components/agent-counters/src/snapshot_test.rs index 2f42aa7708..f54cbc9451 100644 --- a/test-components/agent-counters/src/snapshot_test.rs +++ b/test-components/agent-counters/src/snapshot_test.rs @@ -1,6 +1,12 @@ use golem_rust::{agent_definition, agent_implementation}; use serde::{Deserialize, Serialize}; +#[unsafe(export_name = "_initialize")] +pub extern "C" fn initialize_snapshot_clock() { + // The reactor initializer runs during core instantiation, before snapshot loading. + std::hint::black_box(std::time::Instant::now()); +} + #[agent_definition(snapshotting = "enabled")] trait SnapshotCounter { fn new(id: String) -> Self; From 567c5bd3c8a7610d328eb5d6d187ec264389fc1e Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 08:59:47 +0000 Subject: [PATCH 3/8] Suppress skipped initializer history during snapshot recovery Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-01a08a68-f1d0-7572-8122-6ea73577b108 --- golem-worker-executor/src/worker/mod.rs | 11 ++++++++++- test-components/agent-counters/src/snapshot_test.rs | 5 +---- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/golem-worker-executor/src/worker/mod.rs b/golem-worker-executor/src/worker/mod.rs index 1b9e561538..e508bdba51 100644 --- a/golem-worker-executor/src/worker/mod.rs +++ b/golem-worker-executor/src/worker/mod.rs @@ -7340,7 +7340,7 @@ impl RunningWorker { ); } }; - let context = match Ctx::create( + let mut context = match Ctx::create( worker_metadata.created_by, OwnedAgentId::new(worker_metadata.environment_id, &worker_metadata.agent_id), parent.parsed_agent_id.clone(), @@ -7420,6 +7420,12 @@ impl RunningWorker { ); } }; + if last_snapshot_index.is_some() { + // Core initializers run before load-snapshot, but their recorded host calls are + // already inside the skipped snapshot history. Recreate that runtime state with + // the same durability suppression as snapshot loading, without consuming the tail. + context.begin_call_snapshotting_function(); + } let mut hosted = match instance_host.instantiate(context, &component).await { Ok(hosted) => hosted, Err(error) => { @@ -7435,6 +7441,9 @@ impl RunningWorker { ); } let (instance, mut store) = hosted.into_parts(); + if last_snapshot_index.is_some() { + store.data_mut().end_call_snapshotting_function(); + } if let Some((active_agent, generation)) = entity_generation { let interrupt_state = parent.interrupt_signal.lock().await; if !interrupt_state.has_interrupt() { diff --git a/test-components/agent-counters/src/snapshot_test.rs b/test-components/agent-counters/src/snapshot_test.rs index f54cbc9451..663f990b7c 100644 --- a/test-components/agent-counters/src/snapshot_test.rs +++ b/test-components/agent-counters/src/snapshot_test.rs @@ -22,10 +22,7 @@ struct SnapshotCounterImpl { #[agent_implementation] impl SnapshotCounter for SnapshotCounterImpl { fn new(id: String) -> Self { - Self { - _id: id, - count: 0, - } + Self { _id: id, count: 0 } } fn increment(&mut self) -> u32 { From ceda33a4a830424faa02576baac98bf4a20a6315 Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 09:42:21 +0000 Subject: [PATCH 4/8] Assert snapshot recovery replays an unsnapshotted increment Amp-Thread-ID: https://ampcode.com/threads/T-01a08a68-f1d0-7572-8122-6ea73577b108 Co-authored-by: Amp --- golem-worker-executor/tests/durability.rs | 27 ++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/golem-worker-executor/tests/durability.rs b/golem-worker-executor/tests/durability.rs index cbd8593794..9ff072daa8 100644 --- a/golem-worker-executor/tests/durability.rs +++ b/golem-worker-executor/tests/durability.rs @@ -704,6 +704,12 @@ async fn automatic_snapshot_every_2nd_invocation( .start_agent(&component.id, agent_id.clone()) .await?; + // Construction counts as an invocation; align subsequent snapshots with even increments. + let initial = executor + .invoke_and_await_agent(&component, &agent_id, "get", data_value!()) + .await?; + assert_eq!(initial.into_typed::()?, 0); + for _ in 0..SNAPSHOT_TEST_INVOCATIONS { executor .invoke_and_await_agent(&component, &agent_id, "increment", data_value!()) @@ -728,10 +734,25 @@ async fn automatic_snapshot_every_2nd_invocation( assert_eq!( snapshot_count, - SNAPSHOT_TEST_INVOCATIONS / 2, + 1 + SNAPSHOT_TEST_INVOCATIONS / 2, "Expected a snapshot every 2 invocations" ); + let tail = executor + .invoke_and_await_agent(&component, &agent_id, "increment", data_value!()) + .await?; + assert_eq!(tail.into_typed::()?, 11); + let oplog_with_tail = executor.get_oplog(&worker_id, OplogIndex::INITIAL).await?; + assert_eq!( + oplog_with_tail + .iter() + .rposition(|entry| matches!(&entry.entry, PublicOplogEntry::Snapshot(_))), + oplog + .iter() + .rposition(|entry| matches!(&entry.entry, PublicOplogEntry::Snapshot(_))), + "The final increment must remain outside the last snapshot" + ); + drop(executor); let executor = start_with_snapshot_policy( deps, @@ -748,8 +769,8 @@ async fn automatic_snapshot_every_2nd_invocation( assert_eq!( result_after_restart.into_typed::()?, - SNAPSHOT_TEST_INVOCATIONS as u32, - "Counter should be restored from the automatic snapshot after restart" + 11, + "Counter should include the increment replayed after the automatic snapshot" ); drop(executor); From 9779053a46c1eae06039daa6f09faeeb1fcd779f Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 10:43:09 +0000 Subject: [PATCH 5/8] Wait for poll-loop execution before interrupting workers Amp-Thread-ID: https://ampcode.com/threads/T-01a08ad5-52d3-773d-92fc-a775a25c119f Co-authored-by: Amp --- golem-worker-executor/tests/api.rs | 36 ++++++++++++++++++++++++++---- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/golem-worker-executor/tests/api.rs b/golem-worker-executor/tests/api.rs index a85b581df1..aff4bc9b1a 100644 --- a/golem-worker-executor/tests/api.rs +++ b/golem-worker-executor/tests/api.rs @@ -3877,7 +3877,7 @@ async fn long_running_poll_loop_interrupting_and_resuming_by_second_invocation( .start_agent_with(&component.id, agent_id.clone(), env, Vec::new()) .await?; - executor.log_output(&worker_id).await?; + let (mut rx, _abort_capture) = executor.capture_output_with_termination(&worker_id).await?; executor .invoke_agent(&component, &agent_id, "start_polling", data_value!("first")) @@ -3887,6 +3887,18 @@ async fn long_running_poll_loop_interrupting_and_resuming_by_second_invocation( .wait_for_status(&worker_id, AgentStatus::Running, Duration::from_secs(20)) .await?; + // Running can refer to initialize; interrupting its subsequent idle gap is a no-op. + tokio::time::timeout(Duration::from_secs(30), async { + while let Some(Some(event)) = rx.recv().await { + if stdout_event_matching(&event, "Received initial\n") { + return Ok(()); + } + } + Err(anyhow!("Log stream ended before the first poll completed")) + }) + .await + .map_err(|_| anyhow!("Timed out waiting for poll loop to start"))??; + let values1 = executor .get_running_workers_metadata( &worker_id.component_id, @@ -4364,7 +4376,7 @@ async fn long_running_poll_loop_worker_can_be_deleted_after_interrupt( .start_agent_with(&component.id, agent_id.clone(), env, Vec::new()) .await?; - let (rx, _abort_capture) = executor.capture_output_with_termination(&worker_id).await?; + let (mut rx, _abort_capture) = executor.capture_output_with_termination(&worker_id).await?; executor .invoke_agent(&component, &agent_id, "start_polling", data_value!("first")) @@ -4374,9 +4386,25 @@ async fn long_running_poll_loop_worker_can_be_deleted_after_interrupt( .wait_for_status(&worker_id, AgentStatus::Running, Duration::from_secs(10)) .await?; - executor.interrupt(&worker_id).await?; + // Running can refer to initialize; interrupting its subsequent idle gap is a no-op. + tokio::time::timeout(Duration::from_secs(30), async { + while let Some(Some(event)) = rx.recv().await { + if stdout_event_matching(&event, "Received initial\n") { + return Ok(()); + } + } + Err(anyhow!("Log stream ended before the first poll completed")) + }) + .await + .map_err(|_| anyhow!("Timed out waiting for poll loop to start"))??; - drain_connection(rx).await; + tokio::time::timeout(Duration::from_secs(30), executor.interrupt(&worker_id)) + .await + .map_err(|_| anyhow!("Timed out interrupting poll-loop worker"))??; + + tokio::time::timeout(Duration::from_secs(30), drain_connection(rx)) + .await + .map_err(|_| anyhow!("Timed out waiting for log stream termination after interrupt"))?; executor.check_oplog_is_queryable(&worker_id).await?; executor.delete_worker(&worker_id).await?; From c0ecb1b17b95f9ab8b9d24a97f601c51bf975a11 Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 11:12:28 +0000 Subject: [PATCH 6/8] Align oplog searches with initializer history and await automatic snapshot Amp-Thread-ID: https://ampcode.com/threads/T-01a08a68-f1d0-7572-8122-6ea73577b108 Co-authored-by: Amp --- golem-worker-executor/tests/hot_update.rs | 22 ++++++++++++++------ golem-worker-executor/tests/observability.rs | 3 ++- integration-tests/tests/worker.rs | 3 ++- 3 files changed, 20 insertions(+), 8 deletions(-) diff --git a/golem-worker-executor/tests/hot_update.rs b/golem-worker-executor/tests/hot_update.rs index 2591f3dfb1..b3b7ace585 100644 --- a/golem-worker-executor/tests/hot_update.rs +++ b/golem-worker-executor/tests/hot_update.rs @@ -432,12 +432,22 @@ async fn snapshot_after_auto_update_recovers_with_updated_component_context( .await?; assert_eq!(before_snapshot.into_typed::()?, 0); - let snapshot_count = executor - .get_oplog(&worker_id, OplogIndex::INITIAL) - .await? - .iter() - .filter(|entry| matches!(&entry.entry, PublicOplogEntry::Snapshot(_))) - .count(); + // Automatic snapshot creation is queued after the invocation result is published. + let snapshot_count = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let count = executor + .get_oplog(&worker_id, OplogIndex::INITIAL) + .await? + .iter() + .filter(|entry| matches!(&entry.entry, PublicOplogEntry::Snapshot(_))) + .count(); + if count > snapshots_before_invocation { + break Ok::<_, anyhow::Error>(count); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; assert_eq!(snapshot_count, snapshots_before_invocation + 1); drop(executor); diff --git a/golem-worker-executor/tests/observability.rs b/golem-worker-executor/tests/observability.rs index 79e59c641c..57f7001195 100644 --- a/golem-worker-executor/tests/observability.rs +++ b/golem-worker-executor/tests/observability.rs @@ -216,7 +216,8 @@ async fn search_oplog_1( } assert_eq!(result1.len(), 2, "G1002"); // TODO: this is temporarily not working because of using the dynamic invoke API and not having structured information in the oplog - assert_eq!(result2.len(), 2, "imported-function"); + // Includes the core initializer's monotonic clock call. + assert_eq!(result2.len(), 3, "imported-function"); assert_eq!(result3.len(), 0, "id:G1001 OR id:G1000"); // TODO: this is temporarily not working because of using the dynamic invoke API and not having structured information in the oplog Ok(()) diff --git a/integration-tests/tests/worker.rs b/integration-tests/tests/worker.rs index 152706da3c..3c7e668741 100644 --- a/integration-tests/tests/worker.rs +++ b/integration-tests/tests/worker.rs @@ -1020,7 +1020,8 @@ async fn search_oplog_1(deps: &EnvBasedTestDependencies, _tracing: &Tracing) -> let result3 = user.search_oplog(&agent_id, "G1001 OR G1000").await?; assert_eq!(result1.len(), 2, "G1002"); // TODO: this is temporarily not working because of using the dynamic invoke API and not having structured information in the oplog - assert_eq!(result2.len(), 2, "imported-function"); + // Includes the core initializer's monotonic clock call. + assert_eq!(result2.len(), 3, "imported-function"); assert_eq!(result3.len(), 2, "id:G1001 OR id:G1000"); Ok(()) From cb3ebc86fdcceb42d27c3461d66ba50b5f33584c Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 11:32:33 +0000 Subject: [PATCH 7/8] Wait for durable detach before resuming CLI checkpoint test Amp-Thread-ID: https://ampcode.com/threads/T-01a08ad5-52d3-773d-92fc-a775a25c119f Co-authored-by: Amp --- cli/golem-cli/tests/app/agents.rs | 52 +++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/cli/golem-cli/tests/app/agents.rs b/cli/golem-cli/tests/app/agents.rs index 885d0737db..03ae6b426a 100644 --- a/cli/golem-cli/tests/app/agents.rs +++ b/cli/golem-cli/tests/app/agents.rs @@ -851,23 +851,43 @@ async fn test_streaming_invocation_cli_end_to_end() { .is_some_and(|cursors| !cursors.is_empty()), "checkpoint did not record the item emitted before interruption: {saved_checkpoint}" ); - let resumed = ctx - .cli([ - cmd::AGENT, - cmd::INVOKE, - &resume_agent, - "produce", - &checkpoint_values, - flag::FORMAT, - "json", - "--no-stream", - "--resume-session", - resume_checkpoint.to_str().unwrap(), - ]) - .await; + // Process exit does not wait for the server to persist the transport detach. + let detach_deadline = tokio::time::Instant::now() + Duration::from_secs(10); + let resumed = loop { + let resumed = ctx + .cli([ + cmd::AGENT, + cmd::INVOKE, + &resume_agent, + "produce", + &checkpoint_values, + flag::FORMAT, + "json", + "--no-stream", + "--resume-session", + resume_checkpoint.to_str().unwrap(), + ]) + .await; + if resumed.success() { + break resumed; + } + let events = resumed.stdout_json::(); + if events.len() != 1 + || events[0]["kind"] != "rejected" + || events[0]["reason"] != "invalid-attachment-state" + { + break resumed; + } + assert!( + tokio::time::Instant::now() < detach_deadline, + "the killed invocation session did not become resumable" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + }; assert!( - resumed.success(), - "checkpoint resume failed: {:?}", + resumed.success_or_dump(), + "checkpoint resume failed ({:?}): {:?}", + resumed.status, resumed.stderr().collect::>() ); let resumed_events = resumed From ad937b472493e0f971486a256fde9a435d3ffc3a Mon Sep 17 00:00:00 2001 From: Daniel Vigovszky Date: Thu, 10 Sep 2026 12:12:49 +0000 Subject: [PATCH 8/8] Preserve debugger websocket close reasons in integration failures Amp-Thread-ID: https://ampcode.com/threads/T-01a08a68-f1d0-7572-8122-6ea73577b108 Co-authored-by: Amp --- .../tests/debug_mode/debug_worker_executor.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/golem-debugging-service/tests/debug_mode/debug_worker_executor.rs b/golem-debugging-service/tests/debug_mode/debug_worker_executor.rs index f8ed59b6da..71c3747058 100644 --- a/golem-debugging-service/tests/debug_mode/debug_worker_executor.rs +++ b/golem-debugging-service/tests/debug_mode/debug_worker_executor.rs @@ -92,6 +92,10 @@ impl DebugWorkerExecutorClient { _ => {} } } + Ok(Message::Close(frame)) => { + anyhow::bail!("Debug connection closed: {frame:?}"); + } + Err(error) => return Err(error.into()), _ => { if time.elapsed().as_secs() > 10 { break Err(anyhow::anyhow!("Timeout")); // Break with an error