Skip to content

Stop cloning the agent status record on the invocation path #3830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kmatasfp merged 2 commits into from
Sep 5, 2026
Merged

Stop cloning the agent status record on the invocation path #3830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
15a087a
Read the fields an invocation lookup needs instead of cloning the sta…
kmatasfp Sep 4, 2026
5f8404d
Stop cloning the agent status record on the invocation path
kmatasfp Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions golem-debugging-service/src/debug_context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,9 @@ use golem_common::model::invocation_context::{
self, AttributeValue, InvocationContextStack, SpanId,
};
use golem_common::model::oplog::{AgentError, TimestampedUpdateDescription};
use golem_common::model::regions::DeletedRegions;
use golem_common::model::{
AgentId, AgentInvocation, AgentInvocationOutput, AgentStatusRecord, IdempotencyKey,
OwnedAgentId, Timestamp,
AgentId, AgentInvocation, AgentInvocationOutput, IdempotencyKey, OwnedAgentId, Timestamp,
};
use golem_service_base::error::worker_executor::{InterruptKind, WorkerExecutorError};
use golem_service_base::model::GetFileSystemNodeResult;
Expand Down Expand Up @@ -128,13 +128,13 @@ impl ExternalOperations<Self> for DebugContext {
this: &This,
agent_id: &OwnedAgentId,
agent_mode: AgentMode,
latest_worker_status: &AgentStatusRecord,
deleted_regions: &DeletedRegions,
) -> Option<LastError> {
DurableWorkerCtx::<Self>::get_last_error_and_retry_count(
this,
agent_id,
agent_mode,
latest_worker_status,
deleted_regions,
)
.await
}
Expand Down
10 changes: 5 additions & 5 deletions golem-worker-executor-test-utils/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,12 +50,12 @@ use golem_common::model::oplog::{
TimestampedUpdateDescription, types::ObjectMetadata,
};
use golem_common::model::plan::PlanId;
use golem_common::model::regions::DeletedRegions;
use golem_common::model::retry_policy::NamedRetryPolicy;
use golem_common::model::worker::{AgentConfigEntryDto, AgentMetadataDto};
use golem_common::model::{
AgentFilter, AgentId, AgentInvocation, AgentInvocationOutput, AgentStatusRecord,
IdempotencyKey, OplogIndex, OwnedAgentId, RdbmsPoolKey, RetryConfig, ShardAssignment, ShardId,
TransactionId,
AgentFilter, AgentId, AgentInvocation, AgentInvocationOutput, IdempotencyKey, OplogIndex,
OwnedAgentId, RdbmsPoolKey, RetryConfig, ShardAssignment, ShardId, TransactionId,
};
use golem_service_base::clients::registry::RegistryService;
use golem_service_base::config::{BlobStorageConfig, LocalFileSystemBlobStorageConfig};
Expand Down Expand Up @@ -843,13 +843,13 @@ impl ExternalOperations<TestWorkerCtx> for TestWorkerCtx {
this: &T,
owned_agent_id: &OwnedAgentId,
agent_mode: AgentMode,
latest_worker_status: &AgentStatusRecord,
deleted_regions: &DeletedRegions,
) -> Option<LastError> {
DurableWorkerCtx::<TestWorkerCtx>::get_last_error_and_retry_count(
this,
owned_agent_id,
agent_mode,
latest_worker_status,
deleted_regions,
)
.await
}
Expand Down
29 changes: 15 additions & 14 deletions golem-worker-executor/src/durable_host/durability.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -759,12 +759,12 @@ impl<Ctx: WorkerCtx> InFunctionRetryHost for DurableWorkerCtx<Ctx> {
}

async fn current_retry_state_for(&self, retry_from: OplogIndex) -> Option<RetryPolicyState> {
let latest_status = self
.public_state
self.public_state
.worker()
.get_non_detached_last_known_status()
.await;
latest_status.current_retry_state.get(&retry_from).cloned()
.with_non_detached_last_known_status(|status| {
status.current_retry_state.get(&retry_from).cloned()
})
.await
}

fn durable_execution_state(&self) -> DurableExecutionState {
Expand Down Expand Up @@ -907,16 +907,21 @@ impl<Ctx: WorkerCtx> DurabilityHost for DurableWorkerCtx<Ctx> {
failure: Error,
properties: RetryProperties,
) -> anyhow::Result<()> {
let latest_status = self
.public_state
.worker()
.get_non_detached_last_known_status()
.await;
let current_retry_point = if let Some(region) = self.state.active_atomic_regions.last() {
region.begin_index
} else {
self.state.current_retry_point
};
let current_state = self
.public_state
.worker()
.with_non_detached_last_known_status(|status| {
status
.current_retry_state
.get(&current_retry_point)
.cloned()
})
.await;

// Resolve the matching named retry policy. The synthesized
// default-from-config policy (with `Predicate::True`) is always
Expand Down Expand Up @@ -947,10 +952,6 @@ impl<Ctx: WorkerCtx> DurabilityHost for DurableWorkerCtx<Ctx> {
}
};

let current_state = latest_status
.current_retry_state
.get(&current_retry_point)
.cloned();
let total_attempts = current_state.as_ref().map(|s| s.retry_count()).unwrap_or(0);

match evaluate_named_policy_step_resetting_on_invalid_state(
Expand Down
18 changes: 8 additions & 10 deletions golem-worker-executor/src/durable_host/http/inline_retry.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -711,11 +711,10 @@ pub(crate) fn spawn_http_status_retry_after_body_finish<Ctx: crate::workerctx::W
);

let current_retry_policy_state = worker
.get_non_detached_last_known_status()
.await
.current_retry_state
.get(&begin_index)
.cloned();
.with_non_detached_last_known_status(|status| {
status.current_retry_state.get(&begin_index).cloned()
})
.await;
let mut task_ctx = crate::durable_host::durability::TaskRetryContext {
retry_point: begin_index,
environment_state_service,
Expand Down Expand Up @@ -959,11 +958,10 @@ pub fn spawn_http_request_with_retry<Ctx: crate::workerctx::WorkerCtx>(
Ok(Err(initial_error)) => {
let oplog = worker.oplog();
let current_retry_policy_state = worker
.get_non_detached_last_known_status()
.await
.current_retry_state
.get(&begin_index)
.cloned();
.with_non_detached_last_known_status(|status| {
status.current_retry_state.get(&begin_index).cloned()
})
.await;
let mut task_ctx = crate::durable_host::durability::TaskRetryContext {
retry_point: begin_index,
environment_state_service,
Expand Down
27 changes: 13 additions & 14 deletions golem-worker-executor/src/durable_host/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2522,14 +2522,16 @@ impl<Ctx: WorkerCtx> InvocationHooks for DurableWorkerCtx<Ctx> {

let in_atomic_region = !self.state.active_atomic_regions.is_empty();

let latest_status_before = self
// Only the retry state (one entry per active retry point) is needed here, so copy that
// rather than the whole record.
let current_retry_state_before = self
.public_state
.worker()
.get_non_detached_last_known_status()
.with_non_detached_last_known_status(|status| status.current_retry_state.clone())
.await;
let (decision, retry_policy_state) = self
.get_recovery_decision_on_trap_with_semantic(
&latest_status_before.current_retry_state,
&current_retry_state_before,
trap_type,
in_atomic_region,
full_function_name,
Expand Down Expand Up @@ -2563,10 +2565,10 @@ impl<Ctx: WorkerCtx> InvocationHooks for DurableWorkerCtx<Ctx> {
self.public_state.worker().add_and_commit_oplog(entry).await;
};

let latest_status = self
let latest_agent_status = self
.public_state
.worker()
.get_non_detached_last_known_status()
.with_non_detached_last_known_status(|status| status.status)
.await;

let giving_up = matches!(
Expand All @@ -2576,7 +2578,7 @@ impl<Ctx: WorkerCtx> InvocationHooks for DurableWorkerCtx<Ctx> {
..
}
) || matches!(
latest_status.status,
latest_agent_status,
AgentStatus::Interrupted | AgentStatus::Exited
) || decision == RetryDecision::None;

Expand All @@ -2598,7 +2600,7 @@ impl<Ctx: WorkerCtx> InvocationHooks for DurableWorkerCtx<Ctx> {

debug!(
"Recovery decision for {trap_type:?} with {:?} retries (in_atomic_region={in_atomic_region}): {:?}",
latest_status_before.current_retry_state, decision
current_retry_state_before, decision
);

decision
Expand Down Expand Up @@ -2973,9 +2975,9 @@ impl<Ctx: WorkerCtx> ExternalOperations<Ctx> for DurableWorkerCtx<Ctx> {
this: &T,
owned_agent_id: &OwnedAgentId,
agent_mode: AgentMode,
latest_worker_status: &AgentStatusRecord,
deleted_regions: &DeletedRegions,
) -> Option<LastError> {
last_error(this, owned_agent_id, agent_mode, latest_worker_status).await
last_error(this, owned_agent_id, agent_mode, deleted_regions).await
}

async fn resume_replay(
Expand Down Expand Up @@ -3683,7 +3685,7 @@ async fn last_error<T: HasOplogService + HasConfig>(
this: &T,
owned_agent_id: &OwnedAgentId,
agent_mode: AgentMode,
latest_worker_status: &AgentStatusRecord,
deleted_regions: &DeletedRegions,
) -> Option<LastError> {
let mut idx = this
.oplog_service()
Expand All @@ -3696,10 +3698,7 @@ async fn last_error<T: HasOplogService + HasConfig>(
let mut first_retry_from = OplogIndex::NONE;
let mut last_error_index = idx;
loop {
if latest_worker_status
.deleted_regions
.is_in_deleted_region(idx)
{
if deleted_regions.is_in_deleted_region(idx) {
if idx > OplogIndex::INITIAL {
idx = idx.previous();
continue;
Expand Down
12 changes: 7 additions & 5 deletions golem-worker-executor/src/durable_host/wasm_rpc/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1266,11 +1266,13 @@ fn spawn_rpc_task_with_retry<Ctx: WorkerCtx>(
let execution_status = retry_params.execution_status;
let current_retry_policy_state = retry_params
.worker
.get_non_detached_last_known_status()
.await
.current_retry_state
.get(&retry_params.retry_point)
.cloned();
.with_non_detached_last_known_status(|status| {
status
.current_retry_state
.get(&retry_params.retry_point)
.cloned()
})
.await;
let task_ctx = crate::durable_host::durability::TaskRetryContext {
retry_point: retry_params.retry_point,
environment_state_service: retry_params.environment_state_service,
Expand Down
67 changes: 54 additions & 13 deletions golem-worker-executor/src/grpc/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ use golem_common::model::invocation_context::InvocationContextStack;
use golem_common::model::oplog::types::AgentMetadataForGuests;
use golem_common::model::oplog::{OplogIndex, UpdateDescription};
use golem_common::model::protobuf::to_protobuf_resource_description;
use golem_common::model::regions::DeletedRegions;
use golem_common::model::worker::{AgentConfigEntryDto, AgentMetadataDto, TypedAgentConfigEntry};
use golem_common::model::{
AgentEvent, AgentFilter, AgentFingerprint, AgentId, AgentInvocation, AgentInvocationOutput,
Expand Down Expand Up @@ -181,19 +182,23 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
Ok(worker_executor)
}

/// `agent_status` and `deleted_regions` come from the agent's latest status record. They are
/// taken as fields rather than the record so the invoke path can read them under the status
/// lock instead of copying the record, whose `invocation_results` grows with every invocation.
async fn ensure_not_failed(
&self,
owned_agent_id: &OwnedAgentId,
agent_mode: AgentMode,
metadata: &AgentMetadata,
agent_status: AgentStatus,
deleted_regions: &DeletedRegions,
) -> Result<(), WorkerExecutorError> {
match &metadata.last_known_status.status {
match agent_status {
AgentStatus::Failed => {
let error_and_retry_count = Ctx::get_last_error_and_retry_count(
self,
owned_agent_id,
agent_mode,
&metadata.last_known_status,
deleted_regions,
)
.await;
if let Some(last_error) = error_and_retry_count {
Expand All @@ -214,7 +219,7 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
self,
owned_agent_id,
agent_mode,
&metadata.last_known_status,
deleted_regions,
)
.await;
debug!("Last error and retry count: {:?}", error_and_retry_count);
Expand Down Expand Up @@ -780,8 +785,13 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
owned_agent_id.agent_id(),
))?;

self.ensure_not_failed(&owned_agent_id, metadata.agent_mode, &metadata)
.await?;
self.ensure_not_failed(
&owned_agent_id,
metadata.agent_mode,
metadata.last_known_status.status,
&metadata.last_known_status.deleted_regions,
)
.await?;

match &metadata.last_known_status.status {
AgentStatus::Suspended | AgentStatus::Interrupted | AgentStatus::Idle => {
Expand Down Expand Up @@ -884,10 +894,36 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
.await?;
self.ensure_worker_belongs_to_this_executor(&agent_id)?;

let metadata = Worker::<Ctx>::get_latest_metadata(self, &owned_agent_id).await?;
// This runs on every invocation. For a resident worker, read the two fields the failure
// check needs under the status lock; only a worker that is not resident has its record
// materialised (from the cache and oplog), which is the same cold path as before.
let failure_check =
if let Some(worker) = self.active_workers().try_get(&owned_agent_id).await {
Some(
worker
.with_last_known_status(|status| {
(
worker.agent_mode(),
status.status,
status.deleted_regions.clone(),
)
})
.await,
)
} else {
Worker::<Ctx>::get_latest_metadata(self, &owned_agent_id)
.await?
.map(|metadata| {
(
metadata.agent_mode,
metadata.last_known_status.status,
metadata.last_known_status.deleted_regions,
)
})
};

if let Some(metadata) = &metadata {
self.ensure_not_failed(&owned_agent_id, metadata.agent_mode, metadata)
if let Some((agent_mode, agent_status, deleted_regions)) = failure_check {
self.ensure_not_failed(&owned_agent_id, agent_mode, agent_status, &deleted_regions)
.await?;
}

Expand Down Expand Up @@ -993,7 +1029,7 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
self,
&owned_agent_id,
metadata.agent_mode,
&metadata.last_known_status,
&metadata.last_known_status.deleted_regions,
)
.await;

Expand Down Expand Up @@ -1072,7 +1108,7 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
self,
&worker_metadata.owned_agent_id(),
worker_metadata.agent_mode,
&worker_metadata.last_known_status,
&worker_metadata.last_known_status.deleted_regions,
)
.await;
let metadata =
Expand Down Expand Up @@ -1317,8 +1353,13 @@ impl<Ctx: WorkerCtx, Svcs: HasAll<Ctx> + UsesAllDeps<Ctx = Ctx> + Send + Sync +
owned_agent_id.agent_id(),
))?;

self.ensure_not_failed(&owned_agent_id, metadata.agent_mode, &metadata)
.await?;
self.ensure_not_failed(
&owned_agent_id,
metadata.agent_mode,
metadata.last_known_status.status,
&metadata.last_known_status.deleted_regions,
)
.await?;

if metadata.last_known_status.status != AgentStatus::Interrupted {
let event_service = Worker::get_or_create_suspended(
Expand Down
Loading
Loading