eb1077f380
Source commit: 09ca780bd67a00467e78139cf38466b8201b66ca Public tree identity: sha256:2f744c260b5fc2cf074ad9129f97f87f03714902e5b9fe174128afdc342ec2d0
837 lines
33 KiB
Rust
837 lines
33 KiB
Rust
use std::collections::{BTreeMap, BTreeSet, HashMap};
|
|
use std::io::Read;
|
|
use std::path::PathBuf;
|
|
use std::process::Stdio;
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::thread;
|
|
use std::time::{Duration, Instant};
|
|
|
|
#[cfg(unix)]
|
|
use std::os::unix::process::CommandExt;
|
|
|
|
use crate::coordinator_session::CoordinatorSession;
|
|
use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
|
|
use clusterflux_core::{
|
|
Capability, CommandInvocation, Digest, NodeId, ProcessId, ProjectId, TaskBoundaryValue,
|
|
TaskDispatch, TaskInstanceId, TaskJoinResult, TaskJoinState, TaskSpec, TenantId, VfsManifest,
|
|
WasmExportAbi, WasmHostDebugProbeRequest, WasmHostDebugProbeResult,
|
|
WasmHostSourceSnapshotRequest, WasmHostSourceSnapshotResult, WasmHostTaskHandle,
|
|
WasmHostTaskJoinRequest, WasmHostTaskJoinResult, WasmHostTaskStartRequest,
|
|
WasmHostVfsOperation, WasmHostVfsRequest, WasmHostVfsResult, WasmTaskInvocation,
|
|
WasmTaskOutcome,
|
|
};
|
|
use clusterflux_node::{
|
|
BackendError, CommandOutput, LinuxRootlessPodmanBackend, LocalCheckoutTaskRequest,
|
|
LocalSourceCheckout, PodmanCommand, ProcessOutput, ProcessRunner, WasmDebugControl,
|
|
WasmTaskHost, WasmtimeTaskRuntime,
|
|
};
|
|
|
|
const MAX_CHILD_TASK_HANDLES: usize = 256;
|
|
use crate::daemon::{runtime_task_from_assignment, Args, RuntimeTask};
|
|
use crate::node_identity::signed_node_request_json;
|
|
use crate::source_snapshot::snapshot_project;
|
|
use crate::task_artifacts::{
|
|
retained_result_artifact, task_output_root, NodeArtifactStore, TaskArtifactStore,
|
|
};
|
|
|
|
mod process_runner;
|
|
use process_runner::CoordinatorControlledProcessRunner;
|
|
mod control_watcher;
|
|
use control_watcher::{spawn_task_control_watcher, spawn_worker_shutdown_watcher};
|
|
mod validation;
|
|
use validation::{
|
|
authorize_command_network, bundle_environments, capability_from_descriptor,
|
|
is_secret_environment_name, redact_configured_values, require_command_environment,
|
|
resolve_task_export, task_descriptors, verify_environment_digest, verify_source_snapshot,
|
|
};
|
|
|
|
pub(crate) fn run_verified_wasmtime_assignment(
|
|
args: &Args,
|
|
task: &RuntimeTask,
|
|
node_private_key: &str,
|
|
) -> Result<(CommandOutput, VfsManifest, Option<TaskBoundaryValue>), Box<dyn std::error::Error>> {
|
|
if std::env::var_os("CLUSTERFLUX_DEBUG_CONTROL_TRACE").is_some() {
|
|
eprintln!(
|
|
"clusterflux debug control: starting Wasm assignment for task {}",
|
|
task.task
|
|
);
|
|
}
|
|
let expected_bundle_digest = task.bundle_digest.as_ref().ok_or_else(|| {
|
|
std::io::Error::new(
|
|
std::io::ErrorKind::InvalidInput,
|
|
"wasmtime task assignment includes module bytes but is missing bundle_digest",
|
|
)
|
|
})?;
|
|
let module_base64 = task
|
|
.wasm_module_base64
|
|
.as_ref()
|
|
.expect("caller checked wasm_module_base64");
|
|
let module = BASE64_STANDARD.decode(module_base64)?;
|
|
let actual_bundle_digest = Digest::sha256(&module);
|
|
if &actual_bundle_digest != expected_bundle_digest {
|
|
return Err(format!(
|
|
"bundle digest mismatch: expected {expected_bundle_digest}, received {actual_bundle_digest}"
|
|
)
|
|
.into());
|
|
}
|
|
let task_spec = task.task_spec.as_ref().ok_or_else(|| {
|
|
std::io::Error::new(
|
|
std::io::ErrorKind::InvalidInput,
|
|
"Wasm task assignment is missing its versioned TaskSpec",
|
|
)
|
|
})?;
|
|
let (declared_export, abi) = match &task_spec.dispatch {
|
|
TaskDispatch::CoordinatorNodeWasm { export, abi } => (export.as_deref(), abi),
|
|
};
|
|
let resolved_export;
|
|
let export = match declared_export {
|
|
Some(export) => export,
|
|
None if abi == &WasmExportAbi::TaskV1 => {
|
|
resolved_export = resolve_task_export(&module, task_spec.task_definition.as_str())?;
|
|
&resolved_export
|
|
}
|
|
None => {
|
|
return Err("Wasm entrypoint assignment omitted its descriptor export".into());
|
|
}
|
|
};
|
|
let (stdout, boundary_result) = match abi {
|
|
WasmExportAbi::EntrypointV1 | WasmExportAbi::TaskV1 => {
|
|
let invocation = WasmTaskInvocation::new(
|
|
task_spec.task_definition.clone(),
|
|
task_spec.task_instance.clone(),
|
|
task_spec.args.clone(),
|
|
);
|
|
let result = WasmtimeTaskRuntime::new()?.run_task_export_verified_with_task_host(
|
|
&module,
|
|
expected_bundle_digest,
|
|
export,
|
|
&invocation,
|
|
Box::new(CoordinatorWasmTaskHost::new(
|
|
args,
|
|
task,
|
|
node_private_key,
|
|
&module,
|
|
)?),
|
|
)?;
|
|
if std::env::var_os("CLUSTERFLUX_DEBUG_CONTROL_TRACE").is_some() {
|
|
eprintln!(
|
|
"clusterflux debug control: Wasm assignment returned for task {}",
|
|
task.task
|
|
);
|
|
}
|
|
match result.outcome {
|
|
WasmTaskOutcome::Completed => {
|
|
let boundary = result.result.ok_or("completed Wasm task omitted result")?;
|
|
(
|
|
format!("{}\n", serde_json::to_string(&boundary)?),
|
|
Some(boundary),
|
|
)
|
|
}
|
|
WasmTaskOutcome::Failed => {
|
|
return Err(result
|
|
.error
|
|
.unwrap_or_else(|| "Wasm task failed without an error".to_owned())
|
|
.into())
|
|
}
|
|
}
|
|
}
|
|
};
|
|
let task_id = TaskInstanceId::new(task.task.clone());
|
|
let artifacts = TaskArtifactStore::new(task_id.clone(), NodeId::new(args.node.clone()));
|
|
let manifest = artifacts.flush();
|
|
Ok((
|
|
CommandOutput {
|
|
virtual_thread: task_id,
|
|
status_code: Some(0),
|
|
stdout,
|
|
stderr: String::new(),
|
|
stdout_truncated: false,
|
|
stderr_truncated: false,
|
|
log_backpressured: false,
|
|
staged_artifact: None,
|
|
},
|
|
manifest,
|
|
boundary_result,
|
|
))
|
|
}
|
|
|
|
struct CoordinatorWasmTaskHost {
|
|
args: Args,
|
|
process: String,
|
|
parent_task: String,
|
|
epoch: u64,
|
|
bundle_digest: clusterflux_core::Digest,
|
|
wasm_module_base64: String,
|
|
node_private_key: String,
|
|
allow_command: bool,
|
|
allow_network: bool,
|
|
allow_source_snapshot: bool,
|
|
environment_id: Option<String>,
|
|
environment_digest: Option<Digest>,
|
|
source_snapshot: Option<Digest>,
|
|
output_root: PathBuf,
|
|
task_descriptors: HashMap<String, serde_json::Value>,
|
|
environments: BTreeMap<String, clusterflux_core::EnvironmentResource>,
|
|
next_handle_id: u64,
|
|
handles: Arc<Mutex<HashMap<u64, TaskSpec>>>,
|
|
command_status: Arc<Mutex<Option<String>>>,
|
|
cancellation_requested: Arc<AtomicBool>,
|
|
abort_requested: Arc<AtomicBool>,
|
|
debug_control: Arc<WasmDebugControl>,
|
|
stop_control_watcher: Arc<AtomicBool>,
|
|
}
|
|
|
|
impl CoordinatorWasmTaskHost {
|
|
fn new(
|
|
args: &Args,
|
|
parent: &RuntimeTask,
|
|
node_private_key: &str,
|
|
module: &[u8],
|
|
) -> Result<Self, Box<dyn std::error::Error>> {
|
|
let task_spec = parent
|
|
.task_spec
|
|
.as_ref()
|
|
.ok_or("Wasm task host requires a parent TaskSpec")?;
|
|
let cancellation_requested = Arc::new(AtomicBool::new(false));
|
|
let abort_requested = Arc::new(AtomicBool::new(false));
|
|
let debug_control = Arc::new(WasmDebugControl::default());
|
|
let stop_control_watcher = Arc::new(AtomicBool::new(false));
|
|
let handles = Arc::new(Mutex::new(HashMap::new()));
|
|
let command_status = Arc::new(Mutex::new(None));
|
|
let task_instance = TaskInstanceId::from(parent.task.as_str());
|
|
let output_root =
|
|
task_output_root(args.project_root.as_deref(), &args.node, &task_instance)?;
|
|
let task_args = task_spec
|
|
.args
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(index, value)| (format!("arg_{index}"), format!("{value:?}")))
|
|
.collect();
|
|
spawn_task_control_watcher(
|
|
args.clone(),
|
|
parent.process.clone(),
|
|
parent.task.clone(),
|
|
task_spec.task_definition.as_str().to_owned(),
|
|
node_private_key.to_owned(),
|
|
Arc::clone(&cancellation_requested),
|
|
Arc::clone(&abort_requested),
|
|
Arc::clone(&debug_control),
|
|
task_args,
|
|
Arc::clone(&handles),
|
|
Arc::clone(&command_status),
|
|
Arc::clone(&stop_control_watcher),
|
|
);
|
|
spawn_worker_shutdown_watcher(
|
|
Arc::clone(&abort_requested),
|
|
Arc::clone(&debug_control),
|
|
Arc::clone(&stop_control_watcher),
|
|
);
|
|
Ok(Self {
|
|
args: args.clone(),
|
|
process: parent.process.clone(),
|
|
parent_task: parent.task.clone(),
|
|
epoch: parent.epoch.unwrap_or(task_spec.vfs_epoch),
|
|
bundle_digest: parent
|
|
.bundle_digest
|
|
.clone()
|
|
.ok_or("Wasm task host requires a bundle digest")?,
|
|
wasm_module_base64: BASE64_STANDARD.encode(module),
|
|
node_private_key: node_private_key.to_owned(),
|
|
allow_command: task_spec
|
|
.required_capabilities
|
|
.contains(&Capability::Command)
|
|
&& clusterflux_core::NodeCapabilities::detect_current()
|
|
.capabilities
|
|
.contains(&Capability::Command),
|
|
allow_network: task_spec
|
|
.required_capabilities
|
|
.contains(&Capability::Network)
|
|
&& clusterflux_core::NodeCapabilities::detect_current()
|
|
.capabilities
|
|
.contains(&Capability::Network),
|
|
allow_source_snapshot: task_spec
|
|
.required_capabilities
|
|
.contains(&Capability::SourceFilesystem)
|
|
&& clusterflux_core::NodeCapabilities::detect_current()
|
|
.capabilities
|
|
.contains(&Capability::SourceFilesystem),
|
|
environment_id: task_spec.environment_id.clone(),
|
|
environment_digest: task_spec.environment_digest.clone(),
|
|
source_snapshot: task_spec.source_snapshot.clone(),
|
|
output_root,
|
|
// Wasmtime accepts both binary Wasm and WAT in development tests. Descriptor
|
|
// discovery is required only if the guest actually invokes task_start_v1; module
|
|
// compilation/digest verification remains authoritative for malformed input.
|
|
task_descriptors: task_descriptors(module).unwrap_or_default(),
|
|
environments: bundle_environments(module)?,
|
|
next_handle_id: 1,
|
|
handles,
|
|
command_status,
|
|
cancellation_requested,
|
|
abort_requested,
|
|
debug_control,
|
|
stop_control_watcher,
|
|
})
|
|
}
|
|
|
|
fn session(&self) -> Result<CoordinatorSession, String> {
|
|
CoordinatorSession::connect(&self.args.coordinator).map_err(|error| error.to_string())
|
|
}
|
|
|
|
fn signed_request(
|
|
&self,
|
|
request_kind: &str,
|
|
payload: serde_json::Value,
|
|
) -> Result<serde_json::Value, String> {
|
|
signed_node_request_json(&self.args, &self.node_private_key, request_kind, payload)
|
|
.map_err(|error| error.to_string())
|
|
}
|
|
|
|
fn execute_local_assignment(&self, assignment: &serde_json::Value) -> Result<(), String> {
|
|
let runtime_task =
|
|
runtime_task_from_assignment(assignment).map_err(|error| error.to_string())?;
|
|
let execution =
|
|
run_verified_wasmtime_assignment(&self.args, &runtime_task, &self.node_private_key);
|
|
let (terminal_state, status_code, stdout, stderr, result, retained) = match execution {
|
|
Ok((output, _manifest, result)) => {
|
|
let retained = retained_result_artifact(
|
|
self.args.project_root.as_deref(),
|
|
&self.args.node,
|
|
result.as_ref(),
|
|
);
|
|
match retained {
|
|
Ok(retained) => (
|
|
"completed",
|
|
output.status_code,
|
|
output.stdout,
|
|
output.stderr,
|
|
result,
|
|
retained,
|
|
),
|
|
Err(error) => ("failed", Some(1), String::new(), error, None, None),
|
|
}
|
|
}
|
|
Err(error) => (
|
|
"failed",
|
|
Some(1),
|
|
String::new(),
|
|
error.to_string(),
|
|
None,
|
|
None,
|
|
),
|
|
};
|
|
let artifact_path = retained
|
|
.as_ref()
|
|
.map(|artifact| format!("/vfs/artifacts/{}", artifact.id));
|
|
let artifact_digest = retained.as_ref().map(|artifact| artifact.digest.clone());
|
|
let artifact_size_bytes = retained.as_ref().map(|artifact| artifact.size_bytes);
|
|
let mut session = self.session()?;
|
|
session
|
|
.request(self.signed_request(
|
|
"task_completed",
|
|
serde_json::json!({
|
|
"type": "task_completed",
|
|
"tenant": self.args.tenant,
|
|
"project": self.args.project,
|
|
"process": runtime_task.process,
|
|
"node": self.args.node,
|
|
"task": runtime_task.task,
|
|
"terminal_state": terminal_state,
|
|
"status_code": status_code,
|
|
"stdout_bytes": stdout.len(),
|
|
"stderr_bytes": stderr.len(),
|
|
"stdout_tail": stdout,
|
|
"stderr_tail": stderr,
|
|
"stdout_truncated": false,
|
|
"stderr_truncated": false,
|
|
"artifact_path": artifact_path,
|
|
"artifact_digest": artifact_digest,
|
|
"artifact_size_bytes": artifact_size_bytes,
|
|
"result": result,
|
|
}),
|
|
)?)
|
|
.map_err(|error| error.to_string())?;
|
|
Ok(())
|
|
}
|
|
|
|
fn poll_and_execute_assignment(&self) -> Result<bool, String> {
|
|
let mut session = self.session()?;
|
|
let response = session
|
|
.request(self.signed_request(
|
|
"poll_task_assignment",
|
|
serde_json::json!({
|
|
"type": "poll_task_assignment",
|
|
"tenant": self.args.tenant,
|
|
"project": self.args.project,
|
|
"node": self.args.node,
|
|
}),
|
|
)?)
|
|
.map_err(|error| error.to_string())?;
|
|
let Some(assignment) = response.get("assignment").filter(|value| !value.is_null()) else {
|
|
return Ok(false);
|
|
};
|
|
self.execute_local_assignment(assignment)?;
|
|
Ok(true)
|
|
}
|
|
}
|
|
|
|
impl Drop for CoordinatorWasmTaskHost {
|
|
fn drop(&mut self) {
|
|
self.stop_control_watcher.store(true, Ordering::Release);
|
|
if std::fs::symlink_metadata(&self.output_root)
|
|
.map(|metadata| metadata.is_dir() && !metadata.file_type().is_symlink())
|
|
.unwrap_or(false)
|
|
{
|
|
let _ = std::fs::remove_dir_all(&self.output_root);
|
|
}
|
|
}
|
|
}
|
|
|
|
impl WasmTaskHost for CoordinatorWasmTaskHost {
|
|
fn abort_signal(&self) -> Option<Arc<AtomicBool>> {
|
|
Some(Arc::clone(&self.abort_requested))
|
|
}
|
|
|
|
fn debug_control(&self) -> Option<Arc<WasmDebugControl>> {
|
|
Some(Arc::clone(&self.debug_control))
|
|
}
|
|
|
|
fn start_task(
|
|
&mut self,
|
|
request: WasmHostTaskStartRequest,
|
|
) -> Result<WasmHostTaskHandle, String> {
|
|
request.validate()?;
|
|
let descriptor = self
|
|
.task_descriptors
|
|
.get(request.task_definition.as_str())
|
|
.ok_or_else(|| {
|
|
format!(
|
|
"bundle has no task descriptor named `{}`",
|
|
request.task_definition
|
|
)
|
|
})?;
|
|
let export = descriptor
|
|
.get("export")
|
|
.and_then(serde_json::Value::as_str)
|
|
.filter(|export| !export.trim().is_empty())
|
|
.ok_or_else(|| {
|
|
format!(
|
|
"task `{}` descriptor omitted its Wasm export",
|
|
request.task_definition
|
|
)
|
|
})?;
|
|
let mut required_capabilities = descriptor
|
|
.get("required_capabilities")
|
|
.and_then(serde_json::Value::as_array)
|
|
.into_iter()
|
|
.flatten()
|
|
.map(|value| {
|
|
capability_from_descriptor(
|
|
value
|
|
.as_str()
|
|
.ok_or("task capability descriptor is not a string")?,
|
|
)
|
|
})
|
|
.collect::<Result<BTreeSet<_>, _>>()?;
|
|
let resolved_environment = request
|
|
.environment_id
|
|
.as_deref()
|
|
.map(|name| {
|
|
self.environments.get(name).cloned().ok_or_else(|| {
|
|
format!("bundle environment manifest has no environment `{name}`")
|
|
})
|
|
})
|
|
.transpose()?;
|
|
let environment = resolved_environment
|
|
.as_ref()
|
|
.map(|environment| environment.requirements.clone());
|
|
let environment_digest = resolved_environment
|
|
.as_ref()
|
|
.map(|environment| environment.digest.clone());
|
|
if let Some(environment) = &environment {
|
|
required_capabilities.extend(environment.capabilities.iter().cloned());
|
|
}
|
|
let handle_id = self.next_handle_id;
|
|
let task_instance = clusterflux_core::TaskInstanceId::new(format!(
|
|
"{}:child:{}",
|
|
self.parent_task, handle_id
|
|
));
|
|
let required_artifacts = request
|
|
.args
|
|
.iter()
|
|
.flat_map(TaskBoundaryValue::required_artifacts)
|
|
.collect::<BTreeSet<_>>()
|
|
.into_iter()
|
|
.collect::<Vec<_>>();
|
|
let source_snapshots = request
|
|
.args
|
|
.iter()
|
|
.flat_map(TaskBoundaryValue::source_snapshots)
|
|
.collect::<BTreeSet<_>>();
|
|
if source_snapshots.len() > 1 {
|
|
return Err(
|
|
"one child task invocation cannot require multiple distinct source snapshots"
|
|
.to_owned(),
|
|
);
|
|
}
|
|
if self
|
|
.handles
|
|
.lock()
|
|
.map_err(|_| "Wasm task handle registry lock was poisoned".to_owned())?
|
|
.len()
|
|
>= MAX_CHILD_TASK_HANDLES
|
|
{
|
|
return Err(format!(
|
|
"Wasm child task-handle limit of {MAX_CHILD_TASK_HANDLES} reached"
|
|
));
|
|
}
|
|
let source_snapshot = source_snapshots.into_iter().next();
|
|
let spec = TaskSpec {
|
|
tenant: TenantId::from(self.args.tenant.as_str()),
|
|
project: ProjectId::from(self.args.project.as_str()),
|
|
process: ProcessId::from(self.process.as_str()),
|
|
task_definition: request.task_definition.clone(),
|
|
task_instance: task_instance.clone(),
|
|
dispatch: TaskDispatch::CoordinatorNodeWasm {
|
|
export: Some(export.to_owned()),
|
|
abi: WasmExportAbi::TaskV1,
|
|
},
|
|
environment_id: request.environment_id,
|
|
environment,
|
|
environment_digest,
|
|
required_capabilities,
|
|
dependency_cache: None,
|
|
source_snapshot,
|
|
required_artifacts,
|
|
args: request.args,
|
|
vfs_epoch: self.epoch,
|
|
failure_policy: request.failure_policy,
|
|
bundle_digest: Some(self.bundle_digest.clone()),
|
|
};
|
|
let mut session = self.session()?;
|
|
let response = session
|
|
.request(self.signed_request(
|
|
"launch_child_task",
|
|
serde_json::json!({
|
|
"type": "launch_child_task",
|
|
"tenant": self.args.tenant,
|
|
"project": self.args.project,
|
|
"process": self.process,
|
|
"node": self.args.node,
|
|
"parent_task": self.parent_task,
|
|
"task_spec": spec,
|
|
"wait_for_node": true,
|
|
"artifact_path": format!("/vfs/artifacts/{}-result.json", spec.task_instance),
|
|
"wasm_module_base64": self.wasm_module_base64,
|
|
}),
|
|
)?)
|
|
.map_err(|error| error.to_string())?;
|
|
match response.get("type").and_then(serde_json::Value::as_str) {
|
|
Some("task_launched" | "task_queued") => {}
|
|
other => return Err(format!("unexpected child launch response {other:?}")),
|
|
}
|
|
self.next_handle_id = self.next_handle_id.saturating_add(1);
|
|
self.handles
|
|
.lock()
|
|
.map_err(|_| "Wasm task handle registry lock was poisoned".to_owned())?
|
|
.insert(handle_id, spec.clone());
|
|
Ok(WasmHostTaskHandle {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
handle_id,
|
|
task_spec: spec,
|
|
})
|
|
}
|
|
|
|
fn join_task(
|
|
&mut self,
|
|
request: WasmHostTaskJoinRequest,
|
|
) -> Result<WasmHostTaskJoinResult, String> {
|
|
if request.abi_version != clusterflux_core::WASM_TASK_ABI_VERSION {
|
|
return Err(format!(
|
|
"unsupported Wasm task ABI version {}",
|
|
request.abi_version
|
|
));
|
|
}
|
|
let spec = self
|
|
.handles
|
|
.lock()
|
|
.map_err(|_| "Wasm task handle registry lock was poisoned".to_owned())?
|
|
.get(&request.handle_id)
|
|
.cloned()
|
|
.ok_or_else(|| format!("unknown Wasm task handle {}", request.handle_id))?;
|
|
let started = Instant::now();
|
|
loop {
|
|
let mut session = self.session()?;
|
|
let response = session
|
|
.request(self.signed_request(
|
|
"join_child_task",
|
|
serde_json::json!({
|
|
"type": "join_child_task",
|
|
"tenant": self.args.tenant,
|
|
"project": self.args.project,
|
|
"process": self.process,
|
|
"node": self.args.node,
|
|
"parent_task": self.parent_task,
|
|
"task": spec.task_instance,
|
|
}),
|
|
)?)
|
|
.map_err(|error| error.to_string())?;
|
|
let join: TaskJoinResult = serde_json::from_value(
|
|
response
|
|
.get("join")
|
|
.cloned()
|
|
.ok_or("child task join response omitted join state")?,
|
|
)
|
|
.map_err(|error| error.to_string())?;
|
|
match join.state {
|
|
TaskJoinState::Completed => {
|
|
let result = join
|
|
.result
|
|
.ok_or("completed child task omitted its boundary result")?;
|
|
self.handles
|
|
.lock()
|
|
.map_err(|_| "Wasm task handle registry lock was poisoned".to_owned())?
|
|
.remove(&request.handle_id);
|
|
return Ok(WasmHostTaskJoinResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
task_instance: spec.task_instance,
|
|
result,
|
|
});
|
|
}
|
|
TaskJoinState::Failed | TaskJoinState::Cancelled => {
|
|
self.handles
|
|
.lock()
|
|
.map_err(|_| "Wasm task handle registry lock was poisoned".to_owned())?
|
|
.remove(&request.handle_id);
|
|
return Err(join.message);
|
|
}
|
|
TaskJoinState::Pending => {
|
|
if self.cancellation_requested.load(Ordering::Acquire)
|
|
|| self.abort_requested.load(Ordering::Acquire)
|
|
|| crate::daemon::worker_shutdown_requested()
|
|
{
|
|
return Err(clusterflux_core::limits::TaskJoinError::Cancelled {
|
|
task: spec.task_instance.clone(),
|
|
}
|
|
.to_string());
|
|
}
|
|
let join_timeout = clusterflux_core::limits::task_join_timeout();
|
|
if started.elapsed() > join_timeout {
|
|
return Err(clusterflux_core::limits::TaskJoinError::timeout(
|
|
spec.task_instance.clone(),
|
|
join_timeout,
|
|
)
|
|
.to_string());
|
|
}
|
|
if !self.poll_and_execute_assignment()? {
|
|
thread::sleep(Duration::from_millis(10));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn run_command(
|
|
&mut self,
|
|
request: clusterflux_core::WasmHostCommandRequest,
|
|
) -> Result<clusterflux_core::WasmHostCommandResult, String> {
|
|
request.validate()?;
|
|
if !self.allow_command {
|
|
return Err(
|
|
"Wasm task did not declare Command capability or the selected node did not grant it"
|
|
.to_owned(),
|
|
);
|
|
}
|
|
authorize_command_network(&request.network, self.allow_network)?;
|
|
let environment_id = require_command_environment(self.environment_id.as_deref())?;
|
|
let project_root = self.args.project_root.as_ref().ok_or_else(|| {
|
|
format!(
|
|
"task selected environment `{environment_id}`, but this node has no project checkout or prepared source snapshot; attach it with --project-root"
|
|
)
|
|
})?;
|
|
let expected_snapshot = self.source_snapshot.as_ref().ok_or_else(|| {
|
|
"native command task is missing an explicit SourceSnapshot handle".to_owned()
|
|
})?;
|
|
let actual_snapshot = verify_source_snapshot(project_root, expected_snapshot)?;
|
|
let expected_environment_digest = self.environment_digest.as_ref().ok_or_else(|| {
|
|
format!(
|
|
"task selected environment `{environment_id}` without a bundle-authoritative environment digest"
|
|
)
|
|
})?;
|
|
let environment =
|
|
verify_environment_digest(project_root, environment_id, expected_environment_digest)?;
|
|
let checkout = project_root
|
|
.canonicalize()
|
|
.map_err(|error| format!("resolve node project checkout: {error}"))?;
|
|
let task_id = TaskInstanceId::from(self.parent_task.as_str());
|
|
let mut artifacts =
|
|
TaskArtifactStore::new(task_id.clone(), NodeId::from(self.args.node.as_str()));
|
|
let configured_secrets = request
|
|
.environment_variables
|
|
.iter()
|
|
.filter(|(name, value)| is_secret_environment_name(name) && !value.is_empty())
|
|
.map(|(_, value)| value.clone())
|
|
.collect::<Vec<_>>();
|
|
let mut runner = CoordinatorControlledProcessRunner::new(
|
|
self,
|
|
Duration::from_millis(request.timeout_ms),
|
|
);
|
|
let output = LinuxRootlessPodmanBackend
|
|
.execute_local_checkout_task(
|
|
LocalCheckoutTaskRequest {
|
|
process: ProcessId::from(self.process.as_str()),
|
|
virtual_thread: task_id,
|
|
invocation: &CommandInvocation {
|
|
program: request.program,
|
|
args: request.args,
|
|
working_directory: request.working_directory,
|
|
environment_variables: request.environment_variables,
|
|
timeout_ms: request.timeout_ms,
|
|
network: request.network,
|
|
env: Some(environment),
|
|
},
|
|
checkout: LocalSourceCheckout {
|
|
snapshot: actual_snapshot.digest,
|
|
host_path: checkout,
|
|
},
|
|
output_root: self.output_root.clone(),
|
|
stage_stdout_as: None,
|
|
},
|
|
&mut runner,
|
|
artifacts.overlay_mut(),
|
|
)
|
|
.map_err(|error| {
|
|
if std::env::var_os("CLUSTERFLUX_DEBUG_CONTROL_TRACE").is_some() {
|
|
eprintln!("clusterflux command host failed: {error}");
|
|
}
|
|
error.to_string()
|
|
})?;
|
|
let stdout = redact_configured_values(output.stdout, &configured_secrets);
|
|
let stderr = redact_configured_values(output.stderr, &configured_secrets);
|
|
if std::env::var_os("CLUSTERFLUX_DEBUG_CONTROL_TRACE").is_some() {
|
|
eprintln!(
|
|
"clusterflux command host completed: status={:?} stdout={:?} stderr={:?}",
|
|
output.status_code, stdout, stderr
|
|
);
|
|
}
|
|
Ok(clusterflux_core::WasmHostCommandResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
status_code: output.status_code,
|
|
stdout,
|
|
stderr,
|
|
stdout_truncated: output.stdout_truncated,
|
|
stderr_truncated: output.stderr_truncated,
|
|
})
|
|
}
|
|
|
|
fn poll_task_control(
|
|
&mut self,
|
|
request: clusterflux_core::WasmHostTaskControlRequest,
|
|
) -> Result<clusterflux_core::WasmHostTaskControlResult, String> {
|
|
request.validate()?;
|
|
Ok(clusterflux_core::WasmHostTaskControlResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
cancellation_requested: self.cancellation_requested.load(Ordering::Acquire),
|
|
})
|
|
}
|
|
|
|
fn debug_probe(
|
|
&mut self,
|
|
request: WasmHostDebugProbeRequest,
|
|
) -> Result<WasmHostDebugProbeResult, String> {
|
|
request.validate()?;
|
|
let mut session = self.session()?;
|
|
let response = session
|
|
.request(self.signed_request(
|
|
"report_debug_probe_hit",
|
|
serde_json::json!({
|
|
"type": "report_debug_probe_hit",
|
|
"tenant": self.args.tenant,
|
|
"project": self.args.project,
|
|
"process": self.process,
|
|
"node": self.args.node,
|
|
"task": self.parent_task,
|
|
"probe_symbol": request.symbol,
|
|
}),
|
|
)?)
|
|
.map_err(|error| error.to_string())?;
|
|
if response.get("type").and_then(serde_json::Value::as_str) != Some("debug_probe_hit") {
|
|
return Err(format!(
|
|
"coordinator returned invalid debug probe response: {response}"
|
|
));
|
|
}
|
|
let result = WasmHostDebugProbeResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
breakpoint_matched: response
|
|
.get("breakpoint_matched")
|
|
.and_then(serde_json::Value::as_bool)
|
|
.unwrap_or(false),
|
|
debug_epoch: response
|
|
.get("debug_epoch")
|
|
.and_then(serde_json::Value::as_u64),
|
|
};
|
|
if result.breakpoint_matched {
|
|
if let Some(epoch) = result.debug_epoch {
|
|
self.debug_control.request_freeze(epoch);
|
|
}
|
|
}
|
|
Ok(result)
|
|
}
|
|
|
|
fn vfs_operation(&mut self, request: WasmHostVfsRequest) -> Result<WasmHostVfsResult, String> {
|
|
request.validate()?;
|
|
let store =
|
|
NodeArtifactStore::for_runtime(self.args.project_root.as_deref(), &self.args.node)?;
|
|
let (retained, relative_path) = match request.operation {
|
|
WasmHostVfsOperation::FlushOutput { relative_path } => {
|
|
let retained = store.retain_output_file(&self.output_root, &relative_path)?;
|
|
(retained, relative_path)
|
|
}
|
|
WasmHostVfsOperation::MaterializeArtifact {
|
|
artifact,
|
|
relative_path,
|
|
} => {
|
|
store.materialize_into_output(&artifact, &self.output_root, &relative_path)?;
|
|
let retained = store
|
|
.metadata(&artifact.id)?
|
|
.ok_or_else(|| format!("artifact `{}` became unavailable", artifact.id))?;
|
|
(retained, relative_path)
|
|
}
|
|
};
|
|
Ok(WasmHostVfsResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
artifact: clusterflux_core::ArtifactHandle {
|
|
id: retained.id,
|
|
digest: retained.digest,
|
|
size_bytes: retained.size_bytes,
|
|
},
|
|
relative_path,
|
|
})
|
|
}
|
|
|
|
fn snapshot_source(
|
|
&mut self,
|
|
request: WasmHostSourceSnapshotRequest,
|
|
) -> Result<WasmHostSourceSnapshotResult, String> {
|
|
request.validate()?;
|
|
if !self.allow_source_snapshot {
|
|
return Err(
|
|
"Wasm task did not declare SourceFilesystem capability or the selected node did not grant it"
|
|
.to_owned(),
|
|
);
|
|
}
|
|
let project_root = self.args.project_root.as_deref().ok_or_else(|| {
|
|
"source snapshot task requires a node project checkout; attach it with --project-root"
|
|
.to_owned()
|
|
})?;
|
|
let snapshot = snapshot_project(project_root)?;
|
|
Ok(WasmHostSourceSnapshotResult {
|
|
abi_version: clusterflux_core::WASM_TASK_ABI_VERSION,
|
|
snapshot: snapshot.digest,
|
|
})
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests;
|