diff --git a/Cargo.lock b/Cargo.lock index 2f33499d1a..7fe0a7265d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4245,6 +4245,8 @@ dependencies = [ "async-recursion", "async-scoped", "async-trait", + "aws-config", + "aws-sdk-s3", "axum", "bigdecimal", "bit-vec 0.6.3", @@ -4311,6 +4313,7 @@ dependencies = [ "system-interface", "tempfile", "test-r", + "testcontainers", "tokio", "tokio-stream", "tokio-tungstenite 0.25.0", @@ -4321,6 +4324,7 @@ dependencies = [ "tonic-tracing-opentelemetry", "tracing", "try_match", + "tryhard", "url", "uuid", "wasmtime", diff --git a/cli/golem-cli/src/command_handler/worker/mod.rs b/cli/golem-cli/src/command_handler/worker/mod.rs index 9185b13a54..d24918771a 100644 --- a/cli/golem-cli/src/command_handler/worker/mod.rs +++ b/cli/golem-cli/src/command_handler/worker/mod.rs @@ -61,7 +61,7 @@ use golem_client::api::{AgentClient, ComponentClient, WorkerClient}; use golem_client::model::ScanCursor; use golem_client::model::{ AgentInvocationMode, AgentInvocationRequest, ComponentDto, RevertWorkerTarget, - UpdateWorkerRequest, + UpdateWorkerRequest, WorkersMetadataRequest, }; use golem_common::model::agent::typed_constructor_parameters; use golem_common::model::agent::{AgentConfigSource, AgentMode, AgentTypeName, ParsedAgentId}; @@ -860,6 +860,11 @@ impl WorkerCommandHandler { bail!("Refresh mode is only supported with --format text"); } + let mode_overlay = if mode == AgentListMode::All && !user_set_mode_filter(&filters) { + Some(all_modes_filter()) + } else { + None + }; let filters = apply_list_mode_filter(filters, mode); let (components, filters) = self .resolve_list_components(agent_type_name, component_name, filters) @@ -869,6 +874,7 @@ impl WorkerCommandHandler { self.list_with_refresh( &components, &filters, + mode_overlay.as_ref(), scan_cursor.as_ref(), max_count, precise, @@ -880,6 +886,7 @@ impl WorkerCommandHandler { .list_agents( &components, &filters, + mode_overlay.as_ref(), scan_cursor.as_ref(), max_count, precise, @@ -895,6 +902,7 @@ impl WorkerCommandHandler { &self, components: &[ComponentDto], filters: &[String], + mode_overlay: Option<&AgentFilter>, scan_cursor: Option<&ScanCursor>, max_count: Option, precise: bool, @@ -910,7 +918,15 @@ impl WorkerCommandHandler { // Fetch first — while the previous frame is still visible let output = self - .list_agents(components, filters, scan_cursor, max_count, precise, true) + .list_agents( + components, + filters, + mode_overlay, + scan_cursor, + max_count, + precise, + true, + ) .await .and_then(|view| { self.ctx @@ -1036,6 +1052,7 @@ impl WorkerCommandHandler { &self, components: &[ComponentDto], filters: &[String], + mode_overlay: Option<&AgentFilter>, scan_cursor: Option<&ScanCursor>, max_count: Option, precise: bool, @@ -1066,6 +1083,7 @@ impl WorkerCommandHandler { &component.component_name, &component.id, Some(filters), + mode_overlay, scan_cursor, max_count, precise, @@ -1807,6 +1825,7 @@ impl WorkerCommandHandler { Some(&agent_filters), None, None, + None, false, ) .await?; @@ -2033,7 +2052,7 @@ impl WorkerCommandHandler { component_id: &ComponentId, ) -> anyhow::Result<()> { let (workers, _) = self - .list_component_workers(component_name, component_id, None, None, None, false) + .list_component_workers(component_name, component_id, None, None, None, None, false) .await?; if workers.is_empty() { @@ -2076,7 +2095,7 @@ impl WorkerCommandHandler { show_skip: bool, ) -> anyhow::Result { let (workers, _) = self - .list_component_workers(component_name, component_id, None, None, None, false) + .list_component_workers(component_name, component_id, None, None, None, None, false) .await?; if workers.is_empty() { @@ -2178,6 +2197,7 @@ impl WorkerCommandHandler { component_name: &ComponentName, component_id: &ComponentId, filters: Option<&[String]>, + mode_overlay: Option<&AgentFilter>, start_scan_cursor: Option<&ScanCursor>, max_count: Option, precise: bool, @@ -2186,18 +2206,38 @@ impl WorkerCommandHandler { let mut workers = Vec::::new(); let mut final_result_cursor = Option::::None; - let start_scan_cursor = start_scan_cursor.map(scan_cursor_to_string); - let mut current_scan_cursor = start_scan_cursor.clone(); + // The structured `find_workers_metadata` POST endpoint is used for all + // listing: string filters are converted to a structured `AgentFilter` + // and combined with the optional mode overlay (`--mode all`), which the + // string-filter GET endpoint cannot express as it only supports `And` + // composition. + let string_filter = match filters { + Some(filters) if !filters.is_empty() => Some( + AgentFilter::from(filters.to_vec()) + .map_err(|e| anyhow::anyhow!("Invalid agent filter: {e}"))?, + ), + _ => None, + }; + let filter = match (string_filter, mode_overlay) { + (Some(f), Some(mode_overlay)) => Some(f.and(mode_overlay.clone())), + (Some(f), None) => Some(f), + (None, Some(mode_overlay)) => Some(mode_overlay.clone()), + (None, None) => None, + }; + + let mut current_scan_cursor = start_scan_cursor.cloned(); loop { let result_cursor = { let results = clients .worker - .get_workers_metadata( + .find_workers_metadata( &component_id.0, - filters, - current_scan_cursor.as_deref(), - max_count.or(Some(self.ctx.http_batch_size())), - Some(precise), + &WorkersMetadataRequest { + filter: filter.clone(), + cursor: current_scan_cursor.clone(), + count: max_count.or(Some(self.ctx.http_batch_size())), + precise: Some(precise), + }, ) .await .map_service_error()?; @@ -2215,15 +2255,13 @@ impl WorkerCommandHandler { match result_cursor { Some(next_cursor) => { if max_count.is_none() { - current_scan_cursor = Some(scan_cursor_to_string(&next_cursor)); + current_scan_cursor = Some(next_cursor); } else { final_result_cursor = Some(next_cursor); break; } } - None => { - break; - } + None => break, } } @@ -2931,9 +2969,12 @@ fn secret_config_paths_for_agent_type( /// the user-supplied `--mode` flag, unless the user already provided their own /// `mode ...` filter via `--filter`. /// -/// `AgentListMode::All` never injects a filter (the listing then includes both -/// modes). For `Durable` / `Ephemeral`, the corresponding equality filter is -/// prepended so the executor can use it to scan only the matching mode. +/// `AgentListMode::All` never injects a string filter here; instead +/// [`all_modes_filter`] returns a structured `Or` overlay that the listing +/// path sends via the structured `find_workers_metadata` endpoint so the +/// executor scans both durable and ephemeral oplogs. For `Durable` / +/// `Ephemeral`, the corresponding equality filter is prepended so the +/// executor narrows the oplog scan to only the matching mode. fn apply_list_mode_filter(filters: Vec, mode: AgentListMode) -> Vec { let user_set_mode = filters .iter() @@ -2952,6 +2993,27 @@ fn apply_list_mode_filter(filters: Vec, mode: AgentListMode) -> Vec bool { + filters + .iter() + .any(|s| s.split_whitespace().next().map(str::to_ascii_lowercase) == Some("mode".into())) +} + +/// Structured filter matching agents in either durability mode, used for +/// `--mode all` so the executor scans both durable and ephemeral oplogs +/// (its `modes_from_filter` returns `None` for an `Or` containing `Mode`), +/// while the post-scan matcher accepts both modes. Sent via the structured +/// `find_workers_metadata` endpoint because the string-filter GET endpoint +/// only supports `And` composition. +fn all_modes_filter() -> AgentFilter { + AgentFilter::new_or(vec![ + AgentFilter::new_mode(FilterComparator::Equal, AgentMode::Durable), + AgentFilter::new_mode(FilterComparator::Equal, AgentMode::Ephemeral), + ]) +} + fn parse_worker_error(status: u16, body: Vec) -> ServiceError { let error: anyhow::Result< Option>, diff --git a/cli/golem-cli/tests/app/agents.rs b/cli/golem-cli/tests/app/agents.rs index 64f66e143f..c6348bf3cf 100644 --- a/cli/golem-cli/tests/app/agents.rs +++ b/cli/golem-cli/tests/app/agents.rs @@ -15,6 +15,7 @@ use golem_cli::versions; use indoc::{formatdoc, indoc}; use std::io::Write; use std::path::Path; +use std::time::Duration; use test_r::{inherit_test_dep, test, timeout}; use uuid::Uuid; @@ -1401,6 +1402,372 @@ async fn test_naming_extremes() { assert!(outputs.success_or_dump()); } +// Runs `agent list --mode --format json` against the live server and +// returns the rendered agent-name strings from the response. The JSON view's +// `agentName` is the rendered, language-specific agent id, e.g. +// `DurableListAgent("…")`. Panics on non-zero exit or non-JSON output so test +// failures point at the listing step that broke. +async fn list_agent_names(ctx: &TestContext, mode: &str) -> Vec { + let outputs = ctx + .cli([ + cmd::AGENT, + cmd::LIST, + "--mode", + mode, + flag::FORMAT, + "json", + "--max-count", + "200", + ]) + .await; + assert!( + outputs.success_or_dump(), + "`agent list --mode {mode}` failed" + ); + + let response = outputs + .stdout_json::() + .into_iter() + .next() + .unwrap_or_else(|| panic!("`agent list --mode {mode}` produced no JSON output")); + response.agents.into_iter().map(|a| a.agent_name).collect() +} + +// Scaffolds a fresh Rust app with two agent types: `DurableListAgent` (durable +// by default) and `EphemeralListAgent` (explicitly `mode = "ephemeral"`), then +// builds and deploys it. The caller is responsible for invoking agents so they +// appear in the listing. Returns nothing; the app is live on the started server. +async fn setup_list_mode_filter_app(ctx: &mut TestContext, app_name: &str) { + ctx.start_server().await; + + let outputs = ctx + .cli([flag::YES, cmd::NEW, app_name, flag::TEMPLATE, "rust"]) + .await; + assert!(outputs.success_or_dump()); + + ctx.cd(app_name); + + // Replace the default app manifest with a minimal single-component one, + // dropping the default `httpApi` block so the build does not require a + // deployed HTTP mapping for CounterAgent. + let component_manifest_path = ctx.cwd_path_join("golem.yaml"); + fs::write_str( + &component_manifest_path, + formatdoc! { r#" + manifestVersion: {MANIFEST_VERSION} + + app: {app_name} + + environments: + local: + server: local + componentPresets: debug + cloud: + server: cloud + componentPresets: release + + components: + {app_name}:rust-main: + templates: rust + "#, MANIFEST_VERSION = versions::sdk::MANIFEST }, + ) + .unwrap(); + + // Replace the generated `src/counter_agent.rs` with two agents: one durable + // (the `#[agent_definition]` default) and one explicitly ephemeral via + // `mode = "ephemeral"`. `src/lib.rs` already re-exports `counter_agent::*`. + let component_source_code_file = ctx.cwd_path_join("src/counter_agent.rs"); + fs::write_str( + &component_source_code_file, + indoc! { r#" + use golem_rust::{agent_definition, agent_implementation}; + + #[agent_definition] + pub trait DurableListAgent { + fn new(id: String) -> Self; + fn ping(&self) -> String; + } + + struct DurableListImpl { + id: String, + } + + #[agent_implementation] + impl DurableListAgent for DurableListImpl { + fn new(id: String) -> Self { + Self { id } + } + + fn ping(&self) -> String { + format!("durable:{}", self.id) + } + } + + #[agent_definition(mode = "ephemeral")] + pub trait EphemeralListAgent { + fn new(id: String) -> Self; + fn ping(&self) -> String; + } + + struct EphemeralListImpl { + id: String, + } + + #[agent_implementation] + impl EphemeralListAgent for EphemeralListImpl { + fn new(id: String) -> Self { + Self { id } + } + + fn ping(&self) -> String { + format!("ephemeral:{}", self.id) + } + } + "# }, + ) + .unwrap(); + + let outputs = ctx.cli([cmd::BUILD]).await; + assert!(outputs.success_or_dump()); + + let outputs = ctx.cli([cmd::DEPLOY, flag::YES]).await; + assert!(outputs.success_or_dump()); +} + +// Invokes `agent_type_name("")` `ping` once, creating a worker/oplog entry +// for that agent instance. Panics on non-zero exit or missing ping echo so a +// failure points at the invocation step that broke. +async fn invoke_list_agent(ctx: &TestContext, agent_type_name: &str, id: &str, echo_prefix: &str) { + let outputs = ctx + .cli([ + flag::YES, + cmd::AGENT, + cmd::INVOKE, + &format!("{agent_type_name}(\"{id}\")"), + "ping", + ]) + .await; + assert!(outputs.success_or_dump()); + assert!( + outputs.stdout_contains(format!("{echo_prefix}{id}")), + "invocation of {agent_type_name}(\"{id}\") did not echo {echo_prefix}{id}" + ); +} + +const MODE_FILTER_INSTANCE_COUNT: usize = 20; + +// Verifies that `agent list --mode ephemeral|durable|all` correctly partitions +// the listing by agent durability mode, using a non-trivial number of agent +// instances (20 durable + 20 ephemeral) rather than a single pair. +// +// The CLI injects a `mode == ` metadata filter for `--mode ephemeral` and +// `--mode durable`, while `--mode all` forwards no mode filter so the executor +// scans both modes. The executor's `modes_from_filter` then narrows the oplog +// scan to the requested mode (defaulting to durable when the filter is empty). +// +// Asserts: +// - `--mode ephemeral` lists exactly the 20 ephemeral agents and no durable ones +// - `--mode durable` lists exactly the 20 durable agents and no ephemeral ones +// - `--mode all` lists all 40 agents +// +// Any of these assertions failing surfaces a regression in the +// CLI→worker-service→executor mode-filtering path. +#[test] +#[timeout("15 minutes")] +async fn test_agent_list_mode_filter() { + let mut ctx = TestContext::new(); + let app_name = "agent-list-mode-filter"; + + setup_list_mode_filter_app(&mut ctx, app_name).await; + + // Create 20 durable + 20 ephemeral agent instances. Each invocation with a + // fresh unique id constructs a new agent, and `ping` ensures it gets a + // worker/oplog entry the listing can surface. + let durable_ids: Vec = (0..MODE_FILTER_INSTANCE_COUNT) + .map(|_| Uuid::new_v4().to_string()) + .collect(); + let ephemeral_ids: Vec = (0..MODE_FILTER_INSTANCE_COUNT) + .map(|_| Uuid::new_v4().to_string()) + .collect(); + + for id in &durable_ids { + invoke_list_agent(&ctx, "DurableListAgent", id, "durable:").await; + } + for id in &ephemeral_ids { + invoke_list_agent(&ctx, "EphemeralListAgent", id, "ephemeral:").await; + } + + let count_durable = |names: &[String]| { + names + .iter() + .filter(|n| n.starts_with("DurableListAgent(")) + .count() + }; + let count_ephemeral = |names: &[String]| { + names + .iter() + .filter(|n| n.starts_with("EphemeralListAgent(")) + .count() + }; + + let ephemeral_names = list_agent_names(&ctx, "ephemeral").await; + let e_count = count_ephemeral(&ephemeral_names); + let d_count = count_durable(&ephemeral_names); + assert_eq!( + e_count, MODE_FILTER_INSTANCE_COUNT, + "`agent list --mode ephemeral` must list exactly {MODE_FILTER_INSTANCE_COUNT} ephemeral \ + agents, got {e_count} ephemeral and {d_count} durable: {ephemeral_names:?}" + ); + assert_eq!( + d_count, 0, + "`agent list --mode ephemeral` must not list durable agents, got {d_count}: {ephemeral_names:?}" + ); + + let durable_names = list_agent_names(&ctx, "durable").await; + let d_count = count_durable(&durable_names); + let e_count = count_ephemeral(&durable_names); + assert_eq!( + d_count, MODE_FILTER_INSTANCE_COUNT, + "`agent list --mode durable` must list exactly {MODE_FILTER_INSTANCE_COUNT} durable \ + agents, got {d_count} durable and {e_count} ephemeral: {durable_names:?}" + ); + assert_eq!( + e_count, 0, + "`agent list --mode durable` must not list ephemeral agents, got {e_count}: {durable_names:?}" + ); + + let all_names = list_agent_names(&ctx, "all").await; + let d_count = count_durable(&all_names); + let e_count = count_ephemeral(&all_names); + assert_eq!( + d_count, MODE_FILTER_INSTANCE_COUNT, + "`agent list --mode all` must list exactly {MODE_FILTER_INSTANCE_COUNT} durable agents, \ + got {d_count}: {all_names:?}" + ); + assert_eq!( + e_count, MODE_FILTER_INSTANCE_COUNT, + "`agent list --mode all` must list exactly {MODE_FILTER_INSTANCE_COUNT} ephemeral agents, \ + got {e_count}: {all_names:?}" + ); +} + +// Verifies that the TS REPL `:agent-list --mode ephemeral|durable|all` colon +// command forwards the `--mode` parameter to the underlying `agent list` CLI +// invocation and that the listing is partitioned correctly. +// +// The TS REPL's `:agent-list` command delegates to `golem-cli agent list` via +// the REPL control socket. If the `--mode` flag were silently dropped (e.g. the +// colon-command arg parser failed to forward it), the listing would fall back +// to the `durable` default and never show ephemeral agents. This test catches +// that by asserting `EphemeralListAgent` appears for `--mode ephemeral` and +// `DurableListAgent` appears for `--mode durable`. +// +// Uses a Rust app + TS REPL (the same pattern as `test_rust_counter`) so the +// test does not depend on the TS component template's agent_guest.wasm. +#[test] +#[timeout("10 minutes")] +async fn test_agent_list_mode_filter_in_ts_repl() { + let mut ctx = TestContext::new(); + let app_name = "agent-list-mode-repl"; + + setup_list_mode_filter_app(&mut ctx, app_name).await; + + // Create a small set of durable + ephemeral instances. The REPL test proves + // mode forwarding, not scale — 3 of each is enough to distinguish the + // partitions and keep the interactive session fast. + for _ in 0..3 { + let id = Uuid::new_v4().to_string(); + invoke_list_agent(&ctx, "DurableListAgent", &id, "durable:").await; + } + for _ in 0..3 { + let id = Uuid::new_v4().to_string(); + invoke_list_agent(&ctx, "EphemeralListAgent", &id, "ephemeral:").await; + } + + // The REPL renders `agent list` as a text table in the PTY. Long agent names + // like `EphemeralListAgent("")` wrap across multiple terminal rows in + // the narrow (80-column) PTY, so matching the full name fails. Instead we + // match on the first 8 characters of each agent type name (`Ephemera` / + // `DurableL`) — exactly the width of the Agent name column — which is enough + // to prove the mode filter was forwarded: if `--mode ephemeral` were dropped, + // the default `durable` listing would show `DurableL` and never `Ephemera`. + // + // We don't pass `--format json` to the REPL colon command: the REPL renders + // `agent list` as a text table by default, which is the natural output to + // match here. The short column-width prefixes are enough to prove the mode + // filter was forwarded. + ctx.cli_interactive_repl_test( + [ + cmd::REPL, + flag::LANGUAGE, + "ts", + flag::YES, + "--disable-stream", + ], + move |session| { + session.set_expect_timeout(Some(Duration::from_secs(300))); + session.expect_regex(r"golem-ts-repl\[[^\]]+\]\[[^\]]+\]>")?; + + // The colon-command adapter shows the prompt once or twice after + // each command: once from `displayPrompt()` in the action's finally + // block, and sometimes again from the REPL's own eval callback. We + // always consume the first prompt, then try to consume a second + // with a short timeout. Without consuming at least the first prompt, + // the next `:agent-list` command is not reliably forwarded to the + // server — the REPL's input buffer gets cleared by + // `clearBufferedCommand()` in the action's finally block. + let prompt_regex = r"golem-ts-repl\[[^\]]+\]\[[^\]]+\]>"; + + // `:agent-list --mode ephemeral` must forward the mode and show only + // ephemeral agents. If the mode were dropped, the default `durable` + // listing would show `DurableL` and `Ephemera` would never appear, + // timing out this expectation. + session.send_line_and_expect_regex(r":agent-list --mode ephemeral", r"Ephemera")?; + session.expect_regex(prompt_regex)?; + session.set_expect_timeout(Some(Duration::from_secs(5))); + let _ = session.expect_regex(prompt_regex); + session.set_expect_timeout(Some(Duration::from_secs(300))); + + // `:agent-list --mode durable` must forward the mode and show durable + // agents. + session.send_line_and_expect_regex(r":agent-list --mode durable", r"DurableL")?; + session.expect_regex(prompt_regex)?; + session.set_expect_timeout(Some(Duration::from_secs(5))); + let _ = session.expect_regex(prompt_regex); + session.set_expect_timeout(Some(Duration::from_secs(300))); + + // `:agent-list --mode all` must show both partitions. + session.send_line_and_expect_regex(r":agent-list --mode all", r"DurableL")?; + session.expect_regex(r"Ephemera")?; + session.expect_regex(prompt_regex)?; + session.set_expect_timeout(Some(Duration::from_secs(5))); + let _ = session.expect_regex(prompt_regex); + session.set_expect_timeout(Some(Duration::from_secs(300))); + + session.send_line(".exit")?; + session.expect_eof()?; + + Ok(()) + }, + ) + .await; +} + +// JSON view of the `agent list` structured output. We only need the `agents` +// array and the rendered `agentName` field; serde ignores the `outputType` +// discriminator injected by the CLI's structured-output wrapper. +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct AgentListResponseView { + agents: Vec, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct AgentListView { + agent_name: String, +} + // Use UPDATE_GOLDENFILES=1 or `cargo make cli-integration-tests-update-golden-files` to update files fn check_agent_types_golden_file( application_path: &Path, diff --git a/golem-service-base/src/storage/blob/s3.rs b/golem-service-base/src/storage/blob/s3.rs index 4f065544e6..a402211e51 100644 --- a/golem-service-base/src/storage/blob/s3.rs +++ b/golem-service-base/src/storage/blob/s3.rs @@ -232,6 +232,53 @@ impl S3BlobStorage { Ok(result) } + /// Returns whether any object exists under the given prefix (treated as a + /// directory, i.e. with a trailing `/`). Used to detect implicit directories + /// that have children but no explicit `__dir_marker`. + async fn prefix_has_objects( + &self, + target_label: &'static str, + op_label: &'static str, + bucket: &str, + prefix: &Path, + ) -> Result { + let prefix_str = blob_path_to_string(prefix)?; + let prefix_with_slash = if prefix_str.ends_with('/') { + prefix_str.clone() + } else { + format!("{prefix_str}/") + }; + + let response = with_retries_customized( + target_label, + op_label, + Some(format!("{bucket} - {prefix_str}")), + &self.config.retries, + &(self.client.clone(), bucket, prefix_with_slash), + |(client, bucket, prefix)| { + Box::pin(async move { + client + .list_objects_v2() + .bucket(*bucket) + .prefix(prefix.clone()) + // Only count objects strictly under the prefix, never an + // object whose key is exactly `/` (a degenerate + // self-placeholder), so it is not mistaken for a child. + .start_after(prefix.clone()) + .max_keys(1) + .send() + .await + }) + }, + Self::is_list_objects_v2_error_retriable, + Self::sdk_error_as_loggable_string, + false, + ) + .await?; + + Ok(!response.contents().is_empty()) + } + fn is_get_object_error_retriable( error: &SdkError, ) -> bool { @@ -987,7 +1034,23 @@ impl BlobStorage for S3BlobStorage { Ok(_) => Ok(ExistsResult::Directory), Err(SdkError::ServiceError(service_error)) => { match service_error.into_err() { - HeadObjectError::NotFound(_) => Ok(ExistsResult::DoesNotExist), + HeadObjectError::NotFound(_) => { + // S3 has no real directories: an implicit + // directory can exist (because of nested + // objects or markers) without an explicit + // `__dir_marker` of its own. Match the + // filesystem and in-memory backends by + // reporting a directory whenever any object + // exists under this path's prefix. + if self + .prefix_has_objects(target_label, op_label, bucket, &key) + .await? + { + Ok(ExistsResult::Directory) + } else { + Ok(ExistsResult::DoesNotExist) + } + } err => Err(err.into()), } } diff --git a/golem-worker-executor/Cargo.toml b/golem-worker-executor/Cargo.toml index 9ece01548a..0a33fa5c56 100644 --- a/golem-worker-executor/Cargo.toml +++ b/golem-worker-executor/Cargo.toml @@ -117,6 +117,8 @@ golem-test-framework = { workspace = true } golem-worker-executor-test-utils = { workspace = true } assert2 = { workspace = true } +aws-config = { workspace = true } +aws-sdk-s3 = { workspace = true } criterion = { workspace = true } axum = { workspace = true } futures-concurrency = { workspace = true } @@ -127,5 +129,7 @@ rand = { workspace = true } redis = { workspace = true } serde_json = { workspace = true } test-r = { workspace = true } +testcontainers = { workspace = true } tokio-tungstenite = { workspace = true } +tryhard = { workspace = true } system-interface = { workspace = true } diff --git a/golem-worker-executor/src/services/oplog/tests.rs b/golem-worker-executor/src/services/oplog/tests.rs index 88c23efe3e..24705a6fc9 100644 --- a/golem-worker-executor/src/services/oplog/tests.rs +++ b/golem-worker-executor/src/services/oplog/tests.rs @@ -2651,6 +2651,125 @@ async fn multilayer_scan_for_component(_tracing: &Tracing) { assert_eq!(result.len(), 100); } +/// Ephemeral workers in a multi-layer oplog service live only in the lower +/// (archive) layers - the multi-layer service writes their create entry straight +/// to the first lower layer rather than to the primary. This test verifies that +/// `scan_for_component` discovers such ephemeral workers through the archive-only +/// lower-layer scan path and that mode filtering is honored across layers. +#[test] +async fn multilayer_scan_for_component_ephemeral(_tracing: &Tracing) { + let indexed_storage = Arc::new(InMemoryIndexedStorage::new()); + let blob_storage = Arc::new(InMemoryBlobStorage::new()); + let primary_oplog_service = Arc::new( + PrimaryOplogService::new( + indexed_storage.clone(), + blob_storage.clone(), + 1, + 1, + 100, + RetryConfig::default(), + ) + .await, + ); + let secondary_layer: Arc = Arc::new( + CompressedOplogArchiveService::new(indexed_storage.clone(), 1, RetryConfig::default()), + ); + let tertiary_layer: Arc = + Arc::new(BlobOplogArchiveService::new(blob_storage.clone(), 2)); + + let oplog_service = Arc::new(MultiLayerOplogService::new( + primary_oplog_service.clone(), + nev![secondary_layer.clone(), tertiary_layer.clone()], + // High entry-count limit so no background transfer between lower layers + // happens during the test - the ephemeral workers stay in the first + // lower (archive) layer. + 1000, + 10, + )); + + let account_id = AccountId::new(); + let environment_id = EnvironmentId::new(); + let component_id = ComponentId::new(); + + let create_worker = async |mode: AgentMode, name: String| -> OwnedAgentId { + let agent_id = AgentId { + component_id, + agent_id: name, + }; + let owned_agent_id = OwnedAgentId::new(environment_id, &agent_id); + let create_entry = OplogEntry::create( + agent_id.clone(), + mode, + ComponentRevision::new(1).unwrap(), + Vec::new(), + environment_id, + account_id, + None, + 100, + 100, + HashSet::new(), + Vec::new(), + None, + Uuid::now_v7(), + ); + let oplog = oplog_service + .create( + &owned_agent_id, + mode, + create_entry, + make_agent_metadata(agent_id.clone(), account_id, environment_id), + default_last_known_status(), + default_execution_status(mode), + ) + .await; + oplog.commit(CommitLevel::Always).await; + owned_agent_id + }; + + let mut durable = Vec::new(); + for i in 0..20 { + durable.push(create_worker(AgentMode::Durable, format!("dur-{i}")).await); + } + let mut ephemeral = Vec::new(); + for i in 0..20 { + ephemeral.push(create_worker(AgentMode::Ephemeral, format!("eph-{i}")).await); + } + + // Give any background processes a chance to settle. + tokio::time::sleep(Duration::from_secs(2)).await; + + let drain = async |modes: Option| { + let mut cursor = ScanCursor::default(); + let mut acc: Vec = Vec::new(); + // Use a small page size so pagination crosses layers and mode boundaries. + let page_size = 7; + loop { + let (next_cursor, ids) = oplog_service + .scan_for_component(&environment_id, &component_id, modes, cursor, page_size) + .await + .unwrap(); + acc.extend(ids); + if next_cursor.is_finished() { + break; + } + cursor = next_cursor; + } + acc.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + acc + }; + + let mut expected_durable = durable.clone(); + expected_durable.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + let mut expected_ephemeral = ephemeral.clone(); + expected_ephemeral.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + let mut expected_both: Vec = durable.into_iter().chain(ephemeral).collect(); + expected_both.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + + assert_eq!(drain(Some(AgentMode::Ephemeral)).await, expected_ephemeral); + assert_eq!(drain(Some(AgentMode::Durable)).await, expected_durable); + assert_eq!(drain(None).await, expected_both); +} + /// Reproducer for the oplog unique key violation panic during recovery. /// /// The race is in `OpenOplogs::get_or_open`: when two tasks concurrently call diff --git a/golem-worker-executor/tests/lib.rs b/golem-worker-executor/tests/lib.rs index a9f9d945c0..89c9e1c8b8 100644 --- a/golem-worker-executor/tests/lib.rs +++ b/golem-worker-executor/tests/lib.rs @@ -36,6 +36,7 @@ pub mod key_value_storage; pub mod keyvalue; pub mod namespace_routed_key_value_storage; pub mod observability; +pub mod oplog_blob_archive; pub mod oplog_metrics; pub mod rdbms; pub mod rdbms_service; @@ -109,6 +110,7 @@ tag_suite!(storage_quota, group1); sequential_suite!(key_value_storage); sequential_suite!(namespace_routed_key_value_storage); sequential_suite!(indexed_storage); +sequential_suite!(oplog_blob_archive); timeout_suite!(in_function_retry, "2 minutes"); diff --git a/golem-worker-executor/tests/oplog_blob_archive.rs b/golem-worker-executor/tests/oplog_blob_archive.rs new file mode 100644 index 0000000000..b27b7a1728 --- /dev/null +++ b/golem-worker-executor/tests/oplog_blob_archive.rs @@ -0,0 +1,258 @@ +// Copyright 2024-2026 Golem Cloud +// +// Licensed under the Golem Source License v1.1 (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. + +use async_trait::async_trait; +use aws_config::BehaviorVersion; +use aws_config::meta::region::RegionProviderChain; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::Credentials; +use golem_common::model::agent::AgentMode; +use golem_common::model::component::ComponentId; +use golem_common::model::environment::EnvironmentId; +use golem_common::model::oplog::{LogLevel, OplogEntry, OplogIndex}; +use golem_common::model::{AgentId, OwnedAgentId, ScanCursor}; +use golem_service_base::config::{S3BlobStorageConfig, S3BlobStorageCredentialsConfig}; +use golem_service_base::storage::blob::{BlobStorage, s3}; +use golem_worker_executor::services::oplog::{BlobOplogArchiveService, OplogArchiveService}; +use pretty_assertions::assert_eq; +use std::fmt::Debug; +use std::sync::Arc; +use std::time::Duration; +use test_r::{define_matrix_dimension, test, test_dep}; +use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::runners::AsyncRunner; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; +use tokio::sync::Mutex; + +#[async_trait] +trait GetBlobStorage: Debug + Send + Sync { + async fn get_blob_storage(&self) -> Arc; +} + +struct InMemoryTest; + +impl Debug for InMemoryTest { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "InMemoryTest") + } +} + +#[async_trait] +impl GetBlobStorage for InMemoryTest { + async fn get_blob_storage(&self) -> Arc { + Arc::new(golem_service_base::storage::blob::memory::InMemoryBlobStorage::new()) + } +} + +#[test_dep(scope = PerWorker, tagged_as = "in_memory")] +fn in_memory() -> Arc { + Arc::new(InMemoryTest) +} + +/// Spins up a fresh MinIO container per `get_blob_storage` call and keeps it +/// alive for the lifetime of this (per-worker) dependency, so the returned S3 +/// blob storage remains usable for the whole test. +struct S3Test { + containers: Mutex>>, +} + +impl Debug for S3Test { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "S3Test") + } +} + +#[async_trait] +impl GetBlobStorage for S3Test { + async fn get_blob_storage(&self) -> Arc { + let container = tryhard::retry_fn(|| { + GenericImage::new("minio/minio", "RELEASE.2025-01-20T14-49-07Z") + .with_exposed_port(9000.tcp()) + .with_wait_for(WaitFor::message_on_stderr("API:")) + .with_env_var("MINIO_CONSOLE_ADDRESS", ":9001") + .with_cmd(["server", "/data"]) + .start() + }) + .retries(5) + .exponential_backoff(Duration::from_millis(10)) + .max_delay(Duration::from_secs(10)) + .await + .expect("Failed to start MinIO"); + let host_port = container + .get_host_port_ipv4(9000) + .await + .expect("Failed to get host port"); + + let config = S3BlobStorageConfig { + retries: Default::default(), + region: "us-east-1".to_string(), + object_prefix: String::new(), + aws_endpoint_url: Some(format!("http://127.0.0.1:{host_port}")), + aws_credentials: Some(S3BlobStorageCredentialsConfig::new( + "minioadmin", + "minioadmin", + "test", + )), + ..std::default::Default::default() + }; + create_buckets(host_port, &config).await; + let storage = s3::S3BlobStorage::new(config).await; + + self.containers.lock().await.push(container); + Arc::new(storage) + } +} + +async fn create_buckets(host_port: u16, config: &S3BlobStorageConfig) { + let endpoint_uri = format!("http://127.0.0.1:{host_port}"); + let region_provider = RegionProviderChain::default_provider().or_else("us-east-1"); + let creds = Credentials::new("minioadmin", "minioadmin", None, None, "test"); + let sdk_config = aws_config::defaults(BehaviorVersion::latest()) + .region(region_provider) + .endpoint_url(endpoint_uri) + .credentials_provider(creds) + .load() + .await; + + let client = Client::new(&sdk_config); + for bucket in &config.compressed_oplog_buckets { + client.create_bucket().bucket(bucket).send().await.unwrap(); + } +} + +#[test_dep(scope = PerWorker, tagged_as = "s3")] +fn s3() -> Arc { + Arc::new(S3Test { + containers: Mutex::new(Vec::new()), + }) +} + +define_matrix_dimension!(storage: Arc -> "in_memory", "s3"); + +async fn append_worker( + service: &BlobOplogArchiveService, + owned_agent_id: &OwnedAgentId, + agent_mode: AgentMode, +) { + let archive = service.open(owned_agent_id, agent_mode).await; + archive + .append(vec![( + OplogIndex::INITIAL, + OplogEntry::log(LogLevel::Debug, "test".to_string(), "test".to_string()), + )]) + .await; +} + +async fn drain( + service: &BlobOplogArchiveService, + environment_id: &EnvironmentId, + component_id: &ComponentId, + modes: Option, +) -> Vec { + let mut cursor = ScanCursor::default(); + let mut acc: Vec = Vec::new(); + loop { + let (next_cursor, ids) = service + .scan_for_component(environment_id, component_id, modes, cursor, 100) + .await + .unwrap(); + acc.extend(ids); + if next_cursor.is_finished() { + break; + } + cursor = next_cursor; + } + acc.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + acc +} + +/// The blob oplog archive holds the lowest-layer copies of ephemeral (and +/// archived durable) workers. This test verifies that `scan_for_component` +/// against the blob archive lists workers correctly and filters by agent mode, +/// running the same checks against both an in-memory backend and a real +/// S3-compatible (MinIO) backend. +#[test] +async fn blob_archive_scan_for_component_filters_by_mode( + #[dimension(storage)] storage: &Arc, +) { + let blob_storage = storage.get_blob_storage().await; + + // `compressed_oplog_buckets[0]` ("oplog-archive-1") corresponds to level 0. + let service = BlobOplogArchiveService::new(blob_storage, 0); + + let environment_id = EnvironmentId::new(); + let component_id = ComponentId::new(); + + let make_owned = |name: String| { + OwnedAgentId::new( + environment_id, + &AgentId { + component_id, + agent_id: name, + }, + ) + }; + + let mut ephemeral = Vec::new(); + for i in 0..5 { + let owned = make_owned(format!("eph-{i}")); + append_worker(&service, &owned, AgentMode::Ephemeral).await; + ephemeral.push(owned); + } + let mut durable = Vec::new(); + for i in 0..3 { + let owned = make_owned(format!("dur-{i}")); + append_worker(&service, &owned, AgentMode::Durable).await; + durable.push(owned); + } + + let mut expected_ephemeral = ephemeral.clone(); + expected_ephemeral.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + let mut expected_durable = durable.clone(); + expected_durable.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + let mut expected_both: Vec = ephemeral.into_iter().chain(durable).collect(); + expected_both.sort_by(|a, b| a.agent_id.agent_id.cmp(&b.agent_id.agent_id)); + + assert_eq!( + drain( + &service, + &environment_id, + &component_id, + Some(AgentMode::Ephemeral) + ) + .await, + expected_ephemeral + ); + assert_eq!( + drain( + &service, + &environment_id, + &component_id, + Some(AgentMode::Durable) + ) + .await, + expected_durable + ); + assert_eq!( + drain(&service, &environment_id, &component_id, None).await, + expected_both + ); + + // A component with no archived workers yields an empty scan. + let empty_component_id = ComponentId::new(); + assert_eq!( + drain(&service, &environment_id, &empty_component_id, None).await, + Vec::::new() + ); +}