diff --git a/golem-worker-executor/src/durable_host/call_coordinator.rs b/golem-worker-executor/src/durable_host/call_coordinator.rs index dbde61b8e6..ad6d328ce8 100644 --- a/golem-worker-executor/src/durable_host/call_coordinator.rs +++ b/golem-worker-executor/src/durable_host/call_coordinator.rs @@ -505,7 +505,7 @@ where ctx.state.card_event_boundary_scan = Some(crate::durable_host::CardEventBoundaryScan::new( status.oplog_idx, - status.pending_card_events, + status.pending_card_events.clone(), )); } } diff --git a/golem-worker-executor/src/durable_host/mod.rs b/golem-worker-executor/src/durable_host/mod.rs index f54d73bd1c..96356b8356 100644 --- a/golem-worker-executor/src/durable_host/mod.rs +++ b/golem-worker-executor/src/durable_host/mod.rs @@ -1788,7 +1788,7 @@ impl DurableWorkerCtx { .get_non_detached_last_known_status() .await; let status_idx = status.oplog_idx; - let status_pending = status.pending_card_events; + let status_pending = status.pending_card_events.clone(); let oplog = self.public_state.worker().oplog(); let current_idx = oplog.current_oplog_index().await; diff --git a/golem-worker-executor/src/worker/invocation_loop.rs b/golem-worker-executor/src/worker/invocation_loop.rs index 383d50071f..add3b6c598 100644 --- a/golem-worker-executor/src/worker/invocation_loop.rs +++ b/golem-worker-executor/src/worker/invocation_loop.rs @@ -645,7 +645,8 @@ impl InvocationLoop { .parent .get_non_detached_last_known_status() .await - .current_idempotency_key; + .current_idempotency_key + .clone(); match kind { InterruptKind::Suspend(_) => { self.parent.add_and_commit_oplog(OplogEntry::suspend()).await; diff --git a/golem-worker-executor/src/worker/mod.rs b/golem-worker-executor/src/worker/mod.rs index d3055e15be..2e04e8a60d 100644 --- a/golem-worker-executor/src/worker/mod.rs +++ b/golem-worker-executor/src/worker/mod.rs @@ -1632,7 +1632,7 @@ impl Worker { // Outside of reverts and updates, this will return the same status as get_latest_worker_metadata. // This just has an additional assert built in for when decisions need to be sure that they are fully up to date on the oplog. // _NEVER_ call this from outside the invocation loop, as that is the only place that can reason about whether the status is detached or not. - pub async fn get_non_detached_last_known_status(&self) -> AgentStatusRecord { + pub async fn get_non_detached_last_known_status(&self) -> Arc { // Runs on the worker-state actor's status queue so the detached flag and the published // status are observed consistently with any in-flight commit/reattach transaction. self.state_actor.non_detached_status().await @@ -2428,7 +2428,7 @@ impl Worker { // should only be called from invocation loop pub async fn store_invocation_failure(&self, key: &IdempotencyKey, trap_type: &TrapType) { - let status = self.last_known_status.load_full().as_ref().clone(); + let status = self.last_known_status.load_full(); let keys_to_fail = invocation_keys_to_fail(&status, Some(key), !trap_type.is_invocation_rejection()); let stderr = self.worker_event_service.get_last_invocation_errors(); @@ -5316,7 +5316,12 @@ impl Worker { } pub async fn lookup_invocation_result(&self, key: &IdempotencyKey) -> LookupResult { - let status = self.last_known_status.load_full().as_ref().clone(); + // Kept as an `Arc` rather than cloned out of. The record owns + // `invocation_results`, which gains an entry per invocation and is never + // pruned, so deep-copying it to read one key made each lookup cost more + // than the last. `load_full` already gives a consistent snapshot with the + // lifetime this needs. + let status = self.last_known_status.load_full(); let maybe_result = self .invocation_results .read() @@ -5709,7 +5714,7 @@ impl Worker { } } - let status = self.last_known_status.load_full().as_ref().clone(); + let status = self.last_known_status.load_full(); let keys_to_fail = invocation_keys_to_fail(&status, None, true); let mut invocation_results = self.invocation_results.write().await; @@ -6170,7 +6175,7 @@ impl Worker { if self.last_known_status_detached.load(Ordering::Acquire) { return; } - let status = self.last_known_status.load_full().as_ref().clone(); + let status = self.last_known_status.load_full(); self.status_checkpointer .maybe_checkpoint(&status, reason) .await; @@ -6193,7 +6198,7 @@ impl Worker { if self.last_known_status_detached.load(Ordering::Acquire) { return; } - let status = self.last_known_status.load_full().as_ref().clone(); + let status = self.last_known_status.load_full(); if let Some(marker) = min_exposed_marker && status.oplog_idx > marker { diff --git a/golem-worker-executor/src/worker/state_actor.rs b/golem-worker-executor/src/worker/state_actor.rs index 3dec7c891c..fddb6e0fd5 100644 --- a/golem-worker-executor/src/worker/state_actor.rs +++ b/golem-worker-executor/src/worker/state_actor.rs @@ -129,7 +129,7 @@ enum StatusJob { /// but the caller decides how to react (it asserts): a job whose caller was cancelled must /// not be able to panic the actor. NonDetachedStatus { - done: oneshot::Sender>, + done: oneshot::Sender>>, }, /// Commits, then — if the status became detached (a jump or revert made it non-foldable) — /// recomputes it from the oplog, republishes it, and forces a cache flush. @@ -266,7 +266,7 @@ impl WorkerStateActor { let status = if state.detached.load(Ordering::Acquire) { None } else { - Some(state.last_known_status.load_full().as_ref().clone()) + Some(state.last_known_status.load_full()) }; let _ = done.send(status); } @@ -389,7 +389,7 @@ impl WorkerStateActor { /// Returns the published status, asserting it is attached to the oplog. Serialized behind /// any in-flight commit/reattach transactions. The assert lives here on the caller side, so /// a job left behind by a cancelled caller cannot panic the actor. - pub async fn non_detached_status(&self) -> AgentStatusRecord { + pub async fn non_detached_status(&self) -> Arc { self.commit .run_status_job(|done| StatusJob::NonDetachedStatus { done }) .await