Refactor worker status calculation (#2173)

* refactor worker status computation

* test fixes

* Apply suggestions from code review

Author: mschuwalow

cli/golem-cli/src/model/mod.rs

@@ -463,7 +463,7 @@ pub struct WorkerMetadataView {
     pub env: HashMap<String, String>,
     pub status: golem_client::model::WorkerStatus,
     pub component_version: u64,
-    pub retry_count: u64,
+    pub retry_count: u32,
 
     pub pending_invocation_count: u64,
     pub updates: Vec<golem_client::model::UpdateRecord>,
@@ -516,7 +516,7 @@ pub struct WorkerMetadata {
     pub env: HashMap<String, String>,
     pub status: golem_client::model::WorkerStatus,
     pub component_version: u64,
-    pub retry_count: u64,
+    pub retry_count: u32,
     pub pending_invocation_count: u64,
     pub updates: Vec<golem_client::model::UpdateRecord>,
     pub created_at: DateTime<Utc>,

cli/golem-cli/src/model/text/fmt.rs

@@ -221,7 +221,7 @@ pub fn format_status(status: &WorkerStatus) -> String {
     .to_string()
 }
 
-pub fn format_retry_count(retry_count: &u64) -> String {
+pub fn format_retry_count(retry_count: &u32) -> String {
     if *retry_count == 0 {
         retry_count.to_string()
     } else {

golem-api-grpc/proto/golem/worker/worker_metadata.proto

@@ -19,7 +19,7 @@ message WorkerMetadata {
   map<string, string> env = 4;
   WorkerStatus status = 5;
   uint64 component_version = 6;
-  uint64 retry_count = 7;
+  uint32 retry_count = 7;
   uint64 pending_invocation_count = 8;
   repeated UpdateRecord updates = 9;
   google.protobuf.Timestamp created_at = 10;

golem-common/src/lib.rs

@@ -59,6 +59,8 @@ pub mod retries;
 #[cfg(feature = "serialization")]
 pub mod serialization;
 
+pub mod read_only_lock;
+
 #[cfg(feature = "observability")]
 pub mod tracing;
 

golem-common/src/model/mod.rs

@@ -625,6 +625,7 @@ pub struct WorkerStatusRecord {
     /// The component version at the starting point of the replay. Will be the version of the Create oplog entry
     /// if only automatic updates were used or the version of the latest snapshot-based update
     pub component_version_for_replay: ComponentVersion,
+    pub current_retry_count: u32,
 }
 
 impl<Context> bincode::Decode<Context> for WorkerStatusRecord {
@@ -647,6 +648,7 @@ impl<Context> bincode::Decode<Context> for WorkerStatusRecord {
             active_plugins: Decode::decode(decoder)?,
             deleted_regions: Decode::decode(decoder)?,
             component_version_for_replay: Decode::decode(decoder)?,
+            current_retry_count: Decode::decode(decoder)?,
         })
     }
 }
@@ -672,6 +674,7 @@ impl<'de, Context> BorrowDecode<'de, Context> for WorkerStatusRecord {
             active_plugins: BorrowDecode::borrow_decode(decoder)?,
             deleted_regions: BorrowDecode::borrow_decode(decoder)?,
             component_version_for_replay: BorrowDecode::borrow_decode(decoder)?,
+            current_retry_count: BorrowDecode::borrow_decode(decoder)?,
         })
     }
 }
@@ -696,6 +699,7 @@ impl Default for WorkerStatusRecord {
             active_plugins: HashSet::new(),
             deleted_regions: DeletedRegions::new(),
             component_version_for_replay: 0,
+            current_retry_count: 0,
         }
     }
 }

golem-common/src/model/oplog.rs

@@ -64,7 +64,7 @@ impl OplogIndexRange {
     }
 }
 
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct AtomicOplogIndex(Arc<AtomicU64>);
 
 impl AtomicOplogIndex {

golem-common/src/read_only_lock.rs

@@ -0,0 +1,76 @@
+// Copyright 2024-2025 Golem Cloud
+//
+// Licensed under the Golem Source License v1.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://license.golem.cloud/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod std {
+    use std::sync::Arc;
+    use std::sync::{RwLock, RwLockReadGuard};
+
+    /// Version of std::sync::RwLock that only allows read access to the underlying data.
+    /// Useful if you want to limit which parts of the code are allowed to modify certain data.
+    pub struct ReadOnlyLock<T> {
+        inner: Arc<RwLock<T>>,
+    }
+
+    impl<T> ReadOnlyLock<T> {
+        pub fn new(underlying: Arc<RwLock<T>>) -> Self {
+            Self {
+                inner: underlying.clone(),
+            }
+        }
+
+        pub fn read(&self) -> RwLockReadGuard<'_, T> {
+            self.inner.read().unwrap()
+        }
+    }
+
+    impl<T> Clone for ReadOnlyLock<T> {
+        fn clone(&self) -> Self {
+            Self {
+                inner: self.inner.clone(),
+            }
+        }
+    }
+}
+
+#[cfg(feature = "tokio")]
+pub mod tokio {
+    use std::sync::Arc;
+    use tokio::sync::{RwLock, RwLockReadGuard};
+
+    /// Version of tokio::sync::RwLock that only allows read access to the underlying data.
+    /// Useful if you want to limit which parts of the code are allowed to modify certain data.
+    pub struct ReadOnlyLock<T> {
+        inner: Arc<RwLock<T>>,
+    }
+
+    impl<T> ReadOnlyLock<T> {
+        pub fn new(underlying: Arc<RwLock<T>>) -> Self {
+            Self {
+                inner: underlying.clone(),
+            }
+        }
+
+        pub async fn read(&self) -> RwLockReadGuard<'_, T> {
+            self.inner.read().await
+        }
+    }
+
+    impl<T> Clone for ReadOnlyLock<T> {
+        fn clone(&self) -> Self {
+            Self {
+                inner: self.inner.clone(),
+            }
+        }
+    }
+}

golem-common/src/retries.rs

@@ -26,16 +26,16 @@ use tracing::{error, info, warn, Level};
 /// Returns the delay to be waited before the next retry attempt.
 /// To be called after a failed attempt, with the number of attempts so far.
 /// Returns None if the maximum number of attempts has been reached.
-pub fn get_delay(config: &RetryConfig, attempts: u64) -> Option<Duration> {
+pub fn get_delay(config: &RetryConfig, attempts: u32) -> Option<Duration> {
     // Exponential backoff algorithm inspired by fred::pool::ReconnectPolicy::Exponential
     // Unlike fred, max jitter is not a static value, rather proportional to the current calculated delay
-    if attempts >= (config.max_attempts as u64) {
+    if attempts >= config.max_attempts {
         return None;
     }
 
     let delay_with_opt_jitter = {
         let base_delay = (config.multiplier as u64)
-            .saturating_pow(attempts.saturating_sub(1).try_into().unwrap_or(0))
+            .saturating_pow(attempts.saturating_sub(1))
             .saturating_mul(config.min_delay.as_millis() as u64);
 
         match config.max_jitter_factor {
@@ -61,7 +61,7 @@ pub fn get_delay(config: &RetryConfig, attempts: u64) -> Option<Duration> {
 /// Before attempting to perform the retriable action, call `start_attempt`. If it fails,
 /// call `failed_attempt` and if that returns true, start a new attempt immediately.
 pub struct RetryState<'a> {
-    attempts: u64,
+    attempts: u32,
     retry_config: &'a RetryConfig,
 }
 
@@ -309,7 +309,7 @@ mod tests {
         }
     }
 
-    fn capture_delays(config: &RetryConfig, attempts: &mut u64, delays: &mut Vec<Duration>) {
+    fn capture_delays(config: &RetryConfig, attempts: &mut u32, delays: &mut Vec<Duration>) {
         loop {
             *attempts += 1;
             let delay = super::get_delay(config, *attempts);

golem-debugging-service/src/debug_context.rs

@@ -19,11 +19,10 @@ use golem_common::model::agent::AgentId;
 use golem_common::model::invocation_context::{
     self, AttributeValue, InvocationContextStack, SpanId,
 };
-use golem_common::model::oplog::UpdateDescription;
+use golem_common::model::oplog::{TimestampedUpdateDescription, UpdateDescription};
 use golem_common::model::{
     AccountId, ComponentFilePath, ComponentVersion, GetFileSystemNodeResult, IdempotencyKey,
-    OwnedWorkerId, PluginInstallationId, ProjectId, WorkerId, WorkerMetadata, WorkerStatus,
-    WorkerStatusRecord,
+    OwnedWorkerId, PluginInstallationId, ProjectId, WorkerId, WorkerStatusRecord,
 };
 use golem_service_base::error::worker_executor::{InterruptKind, WorkerExecutorError};
 use golem_wasm_rpc::golem_rpc_0_2_x::types::{
@@ -58,7 +57,7 @@ use golem_worker_executor::services::worker::WorkerService;
 use golem_worker_executor::services::worker_event::WorkerEventService;
 use golem_worker_executor::services::worker_fork::WorkerForkService;
 use golem_worker_executor::services::worker_proxy::WorkerProxy;
-use golem_worker_executor::services::{worker_enumeration, HasAll, HasConfig, HasOplogService};
+use golem_worker_executor::services::{worker_enumeration, HasAll};
 use golem_worker_executor::worker::{RetryDecision, Worker};
 use golem_worker_executor::workerctx::{
     DynamicLinking, ExternalOperations, FileSystemReading, FuelManagement, HasWasiConfigVars,
@@ -121,14 +120,6 @@ impl ExternalOperations<Self> for DebugContext {
         .await
     }
 
-    async fn compute_latest_worker_status<This: HasOplogService + HasConfig + Send + Sync>(
-        this: &This,
-        worker_id: &OwnedWorkerId,
-        metadata: &Option<WorkerMetadata>,
-    ) -> Result<WorkerStatusRecord, WorkerExecutorError> {
-        DurableWorkerCtx::<Self>::compute_latest_worker_status(this, worker_id, metadata).await
-    }
-
     async fn resume_replay(
         store: &mut (impl AsContextMut<Data = Self> + Send),
         instance: &Instance,
@@ -166,23 +157,6 @@ impl ExternalOperations<Self> for DebugContext {
     ) -> Result<(), Error> {
         DurableWorkerCtx::<Self>::on_shard_assignment_changed(this).await
     }
-
-    async fn on_worker_update_failed_to_start<T: HasAll<Self> + Send + Sync>(
-        this: &T,
-        account_id: &AccountId,
-        owned_worker_id: &OwnedWorkerId,
-        target_version: ComponentVersion,
-        details: Option<String>,
-    ) -> Result<(), WorkerExecutorError> {
-        DurableWorkerCtx::<Self>::on_worker_update_failed_to_start(
-            this,
-            account_id,
-            owned_worker_id,
-            target_version,
-            details,
-        )
-        .await
-    }
 }
 
 #[async_trait]
@@ -227,29 +201,13 @@ impl StatusManagement for DebugContext {
         }
     }
 
-    async fn set_suspended(&self) -> Result<(), WorkerExecutorError> {
-        self.durable_ctx.set_suspended().await
+    fn set_suspended(&self) {
+        self.durable_ctx.set_suspended()
     }
 
     fn set_running(&self) {
         self.durable_ctx.set_running()
     }
-
-    async fn get_worker_status(&self) -> WorkerStatus {
-        self.durable_ctx.get_worker_status().await
-    }
-
-    async fn store_worker_status(&self, status: WorkerStatus) {
-        self.durable_ctx.store_worker_status(status).await
-    }
-
-    async fn update_pending_invocations(&self) {
-        self.durable_ctx.update_pending_invocations().await
-    }
-
-    async fn update_pending_updates(&self) {
-        self.durable_ctx.update_pending_updates().await
-    }
 }
 
 #[async_trait]
@@ -588,6 +546,7 @@ impl WorkerCtx for DebugContext {
         project_service: Arc<dyn ProjectService>,
         agent_types_service: Arc<dyn AgentTypesService>,
         shard_service: Arc<dyn ShardService>,
+        pending_update: Option<TimestampedUpdateDescription>,
     ) -> Result<Self, WorkerExecutorError> {
         let golem_ctx = DurableWorkerCtx::create(
             owned_worker_id,
@@ -615,6 +574,7 @@ impl WorkerCtx for DebugContext {
             project_service,
             agent_types_service,
             shard_service,
+            pending_update,
         )
         .await?;
         Ok(Self {

golem-debugging-service/src/oplog/debug_oplog.rs

@@ -16,10 +16,8 @@ use crate::debug_session::{DebugSessionId, DebugSessions};
 use async_trait::async_trait;
 use bytes::Bytes;
 use golem_common::model::oplog::{OplogEntry, OplogIndex, OplogPayload};
-use golem_common::model::WorkerMetadata;
-use golem_worker_executor::model::ExecutionStatus;
 use golem_worker_executor::services::oplog::{CommitLevel, Oplog};
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
@@ -34,14 +32,10 @@ impl DebugOplog {
         inner: Arc<dyn Oplog>,
         debug_session_id: DebugSessionId,
         debug_session: Arc<dyn DebugSessions>,
-        execution_status: Arc<std::sync::RwLock<ExecutionStatus>>,
-        initial_worker_metadata: WorkerMetadata,
     ) -> Self {
         let oplog_state = DebugOplogState {
             debug_session_id,
             debug_session,
-            _execution_status: execution_status,
-            _initial_worker_metadata: initial_worker_metadata,
         };
 
         Self { inner, oplog_state }
@@ -69,8 +63,6 @@ impl Debug for DebugOplog {
 pub struct DebugOplogState {
     debug_session_id: DebugSessionId,
     debug_session: Arc<dyn DebugSessions + Send + Sync>,
-    _execution_status: Arc<std::sync::RwLock<ExecutionStatus>>,
-    _initial_worker_metadata: WorkerMetadata,
 }
 
 #[async_trait]
@@ -82,7 +74,9 @@ impl Oplog for DebugOplog {
     async fn drop_prefix(&self, _last_dropped_id: OplogIndex) {}
 
     // There is no need to commit anything to the indexed storage
-    async fn commit(&self, _level: CommitLevel) {}
+    async fn commit(&self, _level: CommitLevel) -> BTreeMap<OplogIndex, OplogEntry> {
+        BTreeMap::new()
+    }
 
     // Current Oplog Index acts as the Replay Target
     // In a new worker, ReplayState begins with last_replayed_index