golemcloud · vigoo · May 7, 2025 · May 6, 2025 · May 6, 2025 · May 7, 2025
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/golem-test-framework/Cargo.toml b/golem-test-framework/Cargo.toml
@@ -50,6 +50,7 @@ tonic = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
+wasm-metadata = { version = "0.227.1" }
 
 [dev-dependencies]
 test-r = { workspace = true }

diff --git a/golem-test-framework/src/config/cli.rs b/golem-test-framework/src/config/cli.rs
@@ -24,6 +24,7 @@ use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
+use tempfile::TempDir;
 use tracing::{Instrument, Level};
 
 use crate::components::component_compilation_service::docker::DockerComponentCompilationService;
@@ -87,6 +88,7 @@ pub struct CliTestDependencies {
     initial_component_files_service: Arc<InitialComponentFilesService>,
     plugin_wasm_files_service: Arc<PluginWasmFilesService>,
     component_directory: PathBuf,
+    component_temp_directory: Arc<TempDir>,
 }
 
 #[derive(Parser, Debug, Clone)]
@@ -455,6 +457,7 @@ impl CliTestDependencies {
             initial_component_files_service,
             plugin_wasm_files_service,
             component_directory: params.component_directory.clone().into(),
+            component_temp_directory: Arc::new(TempDir::new().unwrap()),
         }
     }
 
@@ -633,6 +636,7 @@ impl CliTestDependencies {
             blob_storage,
             plugin_wasm_files_service,
             initial_component_files_service,
+            component_temp_directory: Arc::new(TempDir::new().unwrap()),
         }
     }
 
@@ -783,6 +787,7 @@ impl CliTestDependencies {
             initial_component_files_service,
             plugin_wasm_files_service,
             component_directory: Path::new(&params.component_directory).to_path_buf(),
+            component_temp_directory: Arc::new(TempDir::new().unwrap()),
         }
     }
 
@@ -946,6 +951,7 @@ impl CliTestDependencies {
             blob_storage,
             plugin_wasm_files_service,
             initial_component_files_service,
+            component_temp_directory: Arc::new(TempDir::new().unwrap()),
         }
     }
 
@@ -1052,6 +1058,7 @@ impl CliTestDependencies {
                     blob_storage,
                     plugin_wasm_files_service,
                     initial_component_files_service,
+                    component_temp_directory: Arc::new(TempDir::new().unwrap()),
                 }
             }
             TestMode::Docker {
@@ -1150,6 +1157,10 @@ impl TestDependencies for CliTestDependencies {
         self.redis.clone()
     }
 
+    fn blob_storage(&self) -> Arc<dyn BlobStorage + Send + Sync + 'static> {
+        self.blob_storage.clone()
+    }
+
     fn redis_monitor(&self) -> Arc<dyn RedisMonitor + Send + Sync + 'static> {
         self.redis_monitor.clone()
     }
@@ -1162,6 +1173,10 @@ impl TestDependencies for CliTestDependencies {
         &self.component_directory
     }
 
+    fn component_temp_directory(&self) -> &Path {
+        self.component_temp_directory.path()
+    }
+
     fn component_service(&self) -> Arc<dyn ComponentService> {
         self.component_service.clone()
     }
@@ -1180,10 +1195,6 @@ impl TestDependencies for CliTestDependencies {
         self.worker_executor_cluster.clone()
     }
 
-    fn blob_storage(&self) -> Arc<dyn BlobStorage + Send + Sync + 'static> {
-        self.blob_storage.clone()
-    }
-
     fn initial_component_files_service(&self) -> Arc<InitialComponentFilesService> {
         self.initial_component_files_service.clone()
     }

diff --git a/golem-test-framework/src/config/env.rs b/golem-test-framework/src/config/env.rs
@@ -46,6 +46,7 @@ use golem_service_base::storage::blob::BlobStorage;
 use std::fmt::{Debug, Formatter};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use tempfile::TempDir;
 use tracing::{Instrument, Level};
 
 pub struct EnvBasedTestDependenciesConfig {
@@ -172,6 +173,7 @@ pub struct EnvBasedTestDependencies {
     blob_storage: Arc<dyn BlobStorage + Send + Sync + 'static>,
     initial_component_files_service: Arc<InitialComponentFilesService>,
     plugin_wasm_files_service: Arc<PluginWasmFilesService>,
+    component_temp_directory: Arc<TempDir>,
 }
 
 impl Debug for EnvBasedTestDependencies {
@@ -491,6 +493,9 @@ impl EnvBasedTestDependencies {
             blob_storage,
             initial_component_files_service,
             plugin_wasm_files_service,
+            component_temp_directory: Arc::new(
+                TempDir::new().expect("Failed to create temporary directory"),
+            ),
         }
     }
 }
@@ -505,6 +510,10 @@ impl TestDependencies for EnvBasedTestDependencies {
         self.redis.clone()
     }
 
+    fn blob_storage(&self) -> Arc<dyn BlobStorage + Send + Sync + 'static> {
+        self.blob_storage.clone()
+    }
+
     fn redis_monitor(&self) -> Arc<dyn RedisMonitor + Send + Sync + 'static> {
         self.redis_monitor.clone()
     }
@@ -517,6 +526,10 @@ impl TestDependencies for EnvBasedTestDependencies {
         &self.config.golem_test_components
     }
 
+    fn component_temp_directory(&self) -> &Path {
+        self.component_temp_directory.path()
+    }
+
     fn component_service(&self) -> Arc<dyn ComponentService> {
         self.component_service.clone()
     }
@@ -535,10 +548,6 @@ impl TestDependencies for EnvBasedTestDependencies {
         self.worker_executor_cluster.clone()
     }
 
-    fn blob_storage(&self) -> Arc<dyn BlobStorage + Send + Sync + 'static> {
-        self.blob_storage.clone()
-    }
-
     fn initial_component_files_service(&self) -> Arc<InitialComponentFilesService> {
         self.initial_component_files_service.clone()
     }

@@ -50,6 +50,7 @@ pub trait TestDependencies {
     fn redis_monitor(&self) -> Arc<dyn RedisMonitor + Send + Sync + 'static>;
     fn shard_manager(&self) -> Arc<dyn ShardManager + Send + Sync + 'static>;
     fn component_directory(&self) -> &Path;
+    fn component_temp_directory(&self) -> &Path;
     fn component_service(&self) -> Arc<dyn ComponentService>;
     fn component_compilation_service(
         &self,

@@ -62,11 +62,13 @@ use golem_wasm_rpc::{Value, ValueAndType};
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::time::{Duration, Instant};
+use tempfile::Builder;
 use tokio::select;
 use tokio::sync::mpsc::UnboundedReceiver;
 use tokio::sync::oneshot::Sender;
 use tracing::{debug, info, Instrument};
 use uuid::Uuid;
+use wasm_metadata::AddMetadata;
 
 pub struct StoreComponentBuilder<'a, DSL: TestDsl + ?Sized> {
     dsl: &'a DSL,
@@ -500,6 +502,17 @@ impl<T: TestDependencies + Send + Sync> TestDsl for T {
                 .map(|(k, v)| (k.to_string(), v.clone())),
         );
 
+        let source_path = if !unverified {
+            rename_component_if_needed(
+                self.component_temp_directory(),
+                &source_path,
+                &component_name,
+            )
+            .expect("Failed to verify and change component metadata")
+        } else {
+            source_path
+        };
+
         let component = {
             if unique {
                 self.component_service()
@@ -2526,3 +2539,35 @@ impl<T: TestDsl + Sync> TestDslUnsafe for T {
         <T as TestDsl>::cancel_invocation(self, worker_id, idempotency_key).await
     }
 }
+
+fn rename_component_if_needed(temp_dir: &Path, path: &Path, name: &str) -> anyhow::Result<PathBuf> {
+    // Check metadata
+    let source = std::fs::read(path)?;
+    let metadata = ComponentMetadata::analyse_component(&source)?;
+    if metadata.root_package_name.is_none() || metadata.root_package_name == Some(name.to_string())
+    {
+        info!(
+            "Name in metadata is {:?}, used component name is {}, using the original WASM: {:?}",
+            metadata.root_package_name, name, path
+        );
+        Ok(path.to_path_buf())
+    } else {
+        let new_path = Builder::new().keep(true).tempfile_in(temp_dir)?;
+        let add_metadata = AddMetadata {
+            name: Some(name.to_string()),
+            version: metadata
+                .root_package_version
+                .map(|v| wasm_metadata::Version::new(v.to_string())),
+            ..Default::default()
+        };
+
+        info!(
+            "Name in metadata is {:?}, used component name is {}, using an updated WASM: {:?}",
+            metadata.root_package_name, name, new_path
+        );
+
+        let updated_wasm = add_metadata.to_wasm(&source)?;
+        std::fs::write(&new_path, updated_wasm)?;
+        Ok(new_path.path().to_path_buf())
+    }
+}
@@ -21,7 +21,9 @@ use crate::model::public_oplog::{
 };
 use crate::model::InterruptKind;
 use crate::preview2::golem_api_1_x;
-use crate::preview2::golem_api_1_x::host::{GetWorkers, Host, HostGetWorkers, WorkerAnyFilter};
+use crate::preview2::golem_api_1_x::host::{
+    ForkResult, GetWorkers, Host, HostGetWorkers, WorkerAnyFilter,
+};
 use crate::preview2::golem_api_1_x::oplog::{
     Host as OplogHost, HostGetOplog, HostSearchOplog, SearchOplog,
 };
@@ -30,6 +32,10 @@ use crate::services::{HasOplogService, HasPlugins, HasWorker};
 use crate::workerctx::{InvocationManagement, StatusManagement, WorkerCtx};
 use anyhow::anyhow;
 use async_trait::async_trait;
+use bincode::de::Decoder;
+use bincode::enc::Encoder;
+use bincode::error::{DecodeError, EncodeError};
+use bincode::Decode;
 use golem_common::model::oplog::{DurableFunctionType, OplogEntry};
 use golem_common::model::regions::OplogRegion;
 use golem_common::model::{ComponentId, ComponentVersion, OwnedWorkerId, ScanCursor, WorkerId};
@@ -729,6 +735,39 @@ impl<Ctx: WorkerCtx> Host for DurableWorkerCtx<Ctx> {
 
         Ok(result.map(|w| w.into()))
     }
+
+    async fn fork(&mut self, new_name: String) -> anyhow::Result<ForkResult> {
+        let durability = Durability::<ForkResult, SerializableError>::new(
+            self,
+            "golem::api",
+            "fork",
+            DurableFunctionType::WriteRemote,
+        )
+        .await?;
+
+        if durability.is_live() {
+            let target_worker_id = WorkerId {
+                component_id: self.owned_worker_id.component_id(),
+                worker_name: new_name.clone(),
+            };
+            let oplog_index_cut_off = self.state.current_oplog_index().await.previous();
+
+            let fork_result = self
+                .state
+                .worker_fork
+                .fork_and_write_fork_result(
+                    &self.owned_worker_id,
+                    &target_worker_id,
+                    oplog_index_cut_off,
+                )
+                .await
+                .map(|_| ForkResult::Original);
+
+            Ok(durability.persist(self, new_name, fork_result).await?)
+        } else {
+            durability.replay(self).await
+        }
+    }
 }
 
 #[async_trait]
@@ -1200,3 +1239,23 @@ impl GetWorkersEntry {
         self.next_cursor = cursor;
     }
 }
+
+impl bincode::Encode for ForkResult {
+    fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
+        match self {
+            ForkResult::Original => bincode::Encode::encode(&0u8, encoder),
+            ForkResult::Forked => bincode::Encode::encode(&1u8, encoder),
+        }
+    }
+}
+
+impl Decode for ForkResult {
+    fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, DecodeError> {
+        let value = <u8 as Decode>::decode(decoder)?;
+        match value {
+            0 => Ok(ForkResult::Original),
+            1 => Ok(ForkResult::Forked),
+            _ => Err(DecodeError::Other("Invalid ForkResult")),
+        }
+    }
+}
@@ -38,6 +38,7 @@ use crate::services::rpc::Rpc;
 use crate::services::scheduler::SchedulerService;
 use crate::services::worker::WorkerService;
 use crate::services::worker_event::WorkerEventService;
+use crate::services::worker_fork::WorkerForkService;
 use crate::services::worker_proxy::WorkerProxy;
 use crate::services::{worker_enumeration, HasAll, HasConfig, HasOplog, HasWorker};
 use crate::services::{HasOplogService, HasPlugins};
@@ -165,6 +166,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
         execution_status: Arc<RwLock<ExecutionStatus>>,
         file_loader: Arc<FileLoader>,
         plugins: Arc<dyn Plugins<Ctx::Types>>,
+        worker_fork: Arc<dyn WorkerForkService + Send + Sync>,
     ) -> Result<Self, GolemError> {
         let temp_dir = Arc::new(tempfile::Builder::new().prefix("golem").tempdir().map_err(
             |e| GolemError::runtime(format!("Failed to create temporary directory: {e}")),
@@ -236,6 +238,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
                 last_oplog_index,
                 component_metadata,
                 worker_config.total_linear_memory_size,
+                worker_fork,
             )
             .await,
             _temp_dir: temp_dir,
@@ -357,6 +360,10 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
         self.state.component_service.clone()
     }
 
+    pub fn worker_fork(&self) -> Arc<dyn WorkerForkService + Send + Sync> {
+        self.state.worker_fork.clone()
+    }
+
     pub fn scheduler_service(&self) -> Arc<dyn SchedulerService + Send + Sync> {
         self.state.scheduler_service.clone()
     }
@@ -2071,6 +2078,8 @@ struct PrivateDurableWorkerState<Ctx: WorkerCtx> {
     invocation_context: InvocationContext,
     current_span_id: SpanId,
     forward_trace_context_headers: bool,
+
+    worker_fork: Arc<dyn WorkerForkService + Send + Sync>,
 }
 
 impl<Ctx: WorkerCtx> PrivateDurableWorkerState<Ctx> {
@@ -2096,6 +2105,7 @@ impl<Ctx: WorkerCtx> PrivateDurableWorkerState<Ctx> {
         last_oplog_index: OplogIndex,
         component_metadata: ComponentMetadata<Ctx::Types>,
         total_linear_memory_size: u64,
+        worker_fork: Arc<dyn WorkerForkService + Send + Sync>,
     ) -> Self {
         let replay_state = ReplayState::new(
             owned_worker_id.clone(),
@@ -2139,6 +2149,7 @@ impl<Ctx: WorkerCtx> PrivateDurableWorkerState<Ctx> {
             invocation_context,
             current_span_id,
             forward_trace_context_headers: true,
+            worker_fork,
         }
     }