use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::io::{BufRead, BufReader, Write}; use std::net::{SocketAddr, TcpListener, TcpStream}; use std::sync::{Arc, Mutex}; use disasmer_core::{ Actor, ArtifactId, ArtifactRegistry, Capability, CapabilityReportError, CredentialKind, DataPlaneObject, DataPlaneScope, DefaultScheduler, Digest, DirectBulkTransferPlan, DownloadLink, DownloadPolicy, EnvironmentRequirements, LimitError, LimitKind, NativeQuicTransport, NodeCapabilities, NodeDescriptor, NodeEndpoint, NodeId, PanelEvent, PanelEventKind, PanelState, PanelWidget, PanelWidgetKind, Placement, PlacementRequest, ProcessId, ProjectId, RateLimit, RendezvousRequest, ResourceLimits, ResourceMeter, Scheduler, SourcePreparation, SourceProviderKind, StorageLocation, TaskId, TenantId, TransportError, UserId, VfsPath, }; use serde::{Deserialize, Serialize}; use thiserror::Error; use crate::{Coordinator, CoordinatorError, InMemoryDurableStore, ProjectRecord}; const MAX_TASK_LOG_TAIL_BYTES: usize = 256 * 1024; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum CoordinatorRequest { Ping, CreateProject { tenant: String, actor_user: String, project: String, name: String, }, SelectProject { tenant: String, actor_user: String, project: String, }, ListProjects { tenant: String, actor_user: String, }, AttachNode { tenant: String, project: String, node: String, public_key: String, }, CreateNodeEnrollmentGrant { tenant: String, project: String, actor_user: String, grant: String, #[serde(default)] now_epoch_seconds: u64, #[serde(default = "default_node_enrollment_ttl_seconds")] ttl_seconds: u64, }, ExchangeNodeEnrollmentGrant { tenant: String, project: String, node: String, public_key: String, enrollment_grant: String, #[serde(default)] now_epoch_seconds: u64, }, NodeHeartbeat { node: String, }, ReportNodeCapabilities { tenant: String, project: String, node: String, capabilities: NodeCapabilities, cached_environment_digests: Vec, #[serde(default)] dependency_cache_digests: Vec, source_snapshots: Vec, artifact_locations: Vec, direct_connectivity: bool, online: bool, }, ListNodeDescriptors { tenant: String, project: String, actor_user: String, }, ScheduleTask { tenant: String, project: String, environment: Option, environment_digest: Option, required_capabilities: Vec, #[serde(default)] dependency_cache: Option, source_snapshot: Option, required_artifacts: Vec, #[serde(default = "default_true")] quota_available: bool, #[serde(default = "default_true")] policy_allowed: bool, prefer_node: Option, }, LaunchTask { tenant: String, project: String, actor_user: String, process: String, task: String, environment: Option, environment_digest: Option, required_capabilities: Vec, #[serde(default)] dependency_cache: Option, source_snapshot: Option, required_artifacts: Vec, #[serde(default = "default_true")] quota_available: bool, #[serde(default = "default_true")] policy_allowed: bool, command: String, #[serde(default)] command_args: Vec, artifact_path: String, }, PollTaskAssignment { tenant: String, project: String, node: String, }, RequestRendezvous { scope: DataPlaneScope, source: NodeEndpoint, destination: NodeEndpoint, direct_connectivity: bool, failure_reason: String, }, RequestSourcePreparation { tenant: String, project: String, provider: SourceProviderKind, }, CompleteSourcePreparation { tenant: String, project: String, node: String, provider: SourceProviderKind, source_snapshot: Digest, }, StartProcess { tenant: String, project: String, process: String, }, ReconnectNode { node: String, process: String, epoch: u64, }, CancelTask { tenant: String, project: String, process: String, node: String, task: String, }, PollTaskControl { tenant: String, project: String, process: String, node: String, task: String, }, PollDebugCommand { tenant: String, project: String, process: String, node: String, task: String, }, ReportTaskLog { tenant: String, project: String, process: String, node: String, task: String, stdout_bytes: u64, stderr_bytes: u64, #[serde(default)] stdout_tail: String, #[serde(default)] stderr_tail: String, stdout_truncated: bool, stderr_truncated: bool, backpressured: bool, }, ReportVfsMetadata { tenant: String, project: String, process: String, node: String, task: String, artifact_path: Option, artifact_digest: Option, artifact_size_bytes: Option, large_bytes_uploaded: bool, }, TaskCompleted { tenant: String, project: String, process: String, node: String, task: String, #[serde(default)] terminal_state: Option, status_code: Option, stdout_bytes: u64, stderr_bytes: u64, #[serde(default)] stdout_tail: String, #[serde(default)] stderr_tail: String, #[serde(default)] stdout_truncated: bool, #[serde(default)] stderr_truncated: bool, artifact_path: Option, artifact_digest: Option, artifact_size_bytes: Option, }, ListTaskEvents { tenant: String, project: String, actor_user: String, #[serde(default)] process: Option, }, RenderOperatorPanel { tenant: String, project: String, process: String, actor_user: String, max_download_bytes: u64, stopped: bool, }, SubmitPanelEvent { tenant: String, project: String, process: String, widget_id: String, kind: PanelEventKind, max_events: u64, }, CreateArtifactDownloadLink { tenant: String, project: String, actor_user: String, artifact: String, max_bytes: u64, token_nonce: String, #[serde(default)] now_epoch_seconds: u64, #[serde(default = "default_download_ttl_seconds")] ttl_seconds: u64, }, OpenArtifactDownloadStream { tenant: String, project: String, actor_user: String, artifact: String, max_bytes: u64, token_nonce: String, token_digest: Digest, #[serde(default)] now_epoch_seconds: u64, chunk_bytes: u64, }, RevokeArtifactDownloadLink { tenant: String, project: String, actor_user: String, artifact: String, token_digest: Digest, }, ExportArtifactToNode { tenant: String, project: String, actor_user: String, artifact: String, receiver_node: String, direct_connectivity: bool, failure_reason: String, }, } type TaskControlKey = (TenantId, ProjectId, ProcessId, NodeId, TaskId); type TaskAssignmentKey = (TenantId, ProjectId, NodeId); type PanelStopKey = (TenantId, ProjectId, ProcessId); type EnrollmentGrantKey = (TenantId, ProjectId, String); fn default_download_ttl_seconds() -> u64 { 900 } fn default_node_enrollment_ttl_seconds() -> u64 { 900 } fn default_true() -> bool { true } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TaskTerminalState { Completed, Failed, Cancelled, } impl TaskTerminalState { fn from_status_code(status_code: Option) -> Self { match status_code { Some(0) => Self::Completed, _ => Self::Failed, } } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct TaskCompletionEvent { pub tenant: TenantId, pub project: ProjectId, pub process: ProcessId, pub node: NodeId, pub task: TaskId, pub terminal_state: TaskTerminalState, pub status_code: Option, pub stdout_bytes: u64, pub stderr_bytes: u64, pub stdout_tail: String, pub stderr_tail: String, pub stdout_truncated: bool, pub stderr_truncated: bool, pub artifact_path: Option, pub artifact_digest: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct TaskAssignment { pub tenant: TenantId, pub project: ProjectId, pub process: ProcessId, pub task: TaskId, pub node: NodeId, pub epoch: u64, pub command: String, pub command_args: Vec, pub artifact_path: String, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum SourcePreparationDisposition { Pending { reason: String }, Assigned { node: NodeId }, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SourcePreparationStatus { pub preparation: SourcePreparation, pub disposition: SourcePreparationDisposition, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum CoordinatorResponse { Pong { epoch: u64, }, ProjectCreated { project: ProjectRecord, actor: UserId, }, ProjectSelected { project: ProjectRecord, actor: UserId, }, Projects { projects: Vec, actor: UserId, }, NodeAttached { node: NodeId, tenant: TenantId, project: ProjectId, }, NodeEnrollmentGrantCreated { tenant: TenantId, project: ProjectId, grant: String, scope: String, expires_at_epoch_seconds: u64, }, NodeEnrollmentExchanged { node: NodeId, tenant: TenantId, project: ProjectId, credential: disasmer_core::NodeCredential, }, NodeHeartbeat { node: NodeId, epoch: u64, }, NodeCapabilitiesRecorded { node: NodeId, node_descriptors: usize, }, NodeDescriptors { descriptors: Vec, actor: UserId, }, TaskPlacement { placement: Placement, }, TaskLaunched { process: ProcessId, task: TaskId, placement: Placement, assignment: TaskAssignment, }, TaskAssignment { assignment: Option, }, RendezvousPlan { plan: DirectBulkTransferPlan, charged_rendezvous_attempts: u64, }, SourcePreparation { status: SourcePreparationStatus, }, SourcePreparationCompleted { node: NodeId, provider: SourceProviderKind, source_snapshot: Digest, }, ProcessStarted { process: ProcessId, epoch: u64, }, NodeReconnected { node: NodeId, process: ProcessId, }, TaskCancellationRequested { process: ProcessId, task: TaskId, node: NodeId, }, TaskControl { process: ProcessId, task: TaskId, cancel_requested: bool, }, DebugCommand { process: ProcessId, task: TaskId, command: Option, }, TaskLogRecorded { process: ProcessId, task: TaskId, stdout_bytes: u64, stderr_bytes: u64, stdout_tail: String, stderr_tail: String, backpressured: bool, }, VfsMetadataRecorded { process: ProcessId, task: TaskId, artifact_path: Option, large_bytes_uploaded: bool, }, TaskRecorded { process: ProcessId, task: TaskId, events_recorded: usize, }, TaskEvents { events: Vec, }, OperatorPanel { panel: PanelState, }, PanelEventAccepted { used_events: u64, max_events: u64, }, ArtifactDownloadLink { link: DownloadLink, }, ArtifactDownloadLinkRevoked { link: DownloadLink, }, ArtifactDownloadStream { link: DownloadLink, streamed_bytes: u64, charged_download_bytes: u64, }, ArtifactExportPlan { plan: DirectBulkTransferPlan, source_node: NodeId, receiver_node: NodeId, }, Error { message: String, }, } #[derive(Debug, Error)] pub enum CoordinatorServiceError { #[error("coordinator protocol I/O error: {0}")] Io(#[from] std::io::Error), #[error("coordinator protocol JSON error: {0}")] Json(#[from] serde_json::Error), #[error("coordinator request failed: {0}")] Coordinator(#[from] CoordinatorError), #[error("artifact download request failed: {0}")] Download(#[from] disasmer_core::DownloadError), #[error("scheduler placement failed: {0}")] Scheduler(#[from] disasmer_core::PlacementError), #[error("transport request failed: {0}")] Transport(#[from] TransportError), #[error("resource limit failed: {0}")] Resource(#[from] LimitError), #[error("operator panel request failed: {0}")] Panel(#[from] disasmer_core::PanelError), #[error("invalid node capability report: {0}")] CapabilityReport(#[from] CapabilityReportError), #[error("invalid VFS artifact path reported by node: {0}")] InvalidArtifactPath(String), #[error("invalid task log tail reported by node: {0}")] InvalidTaskLogTail(String), } #[derive(Clone, Debug)] pub struct CoordinatorService { coordinator: Coordinator, store: InMemoryDurableStore, node_descriptors: BTreeMap, enrollment_grants: BTreeMap, task_events: Vec, task_assignments: BTreeMap>, task_cancellations: BTreeSet, panel_snapshots: BTreeMap, stopped_panels: BTreeSet, panel_event_limits: BTreeMap<(TenantId, ProjectId, ProcessId, String), RateLimit>, artifact_registry: ArtifactRegistry, transport: NativeQuicTransport, rendezvous_limits: ResourceLimits, rendezvous_meter: ResourceMeter, download_limits: ResourceLimits, download_meter: ResourceMeter, } impl CoordinatorService { pub fn new(coordinator_epoch: u64) -> Self { let store = InMemoryDurableStore::default(); let coordinator = Coordinator::boot(&store, coordinator_epoch); Self { coordinator, store, node_descriptors: BTreeMap::new(), enrollment_grants: BTreeMap::new(), task_events: Vec::new(), task_assignments: BTreeMap::new(), task_cancellations: BTreeSet::new(), panel_snapshots: BTreeMap::new(), stopped_panels: BTreeSet::new(), panel_event_limits: BTreeMap::new(), artifact_registry: ArtifactRegistry::default(), transport: NativeQuicTransport, rendezvous_limits: ResourceLimits::community_tier_defaults(), rendezvous_meter: ResourceMeter::default(), download_limits: ResourceLimits::community_tier_defaults(), download_meter: ResourceMeter::default(), } } pub fn handle_request( &mut self, request: CoordinatorRequest, ) -> Result { match request { CoordinatorRequest::Ping => Ok(CoordinatorResponse::Pong { epoch: self.coordinator.coordinator_epoch(), }), CoordinatorRequest::CreateProject { tenant, actor_user, project, name, } => { let tenant = TenantId::new(tenant); let actor = UserId::new(actor_user); let project = ProjectId::new(project); if let Some(existing) = self.coordinator.project(&project) { if existing.tenant != tenant { return Err(CoordinatorError::Unauthorized( "project id is outside the signed-in tenant scope".to_owned(), ) .into()); } } self.coordinator.upsert_tenant(tenant.clone()); self.coordinator.upsert_user( tenant.clone(), actor.clone(), CredentialKind::BrowserSession, ); self.coordinator .upsert_project(tenant.clone(), project.clone(), name); self.coordinator.grant_project_debug( tenant.clone(), project.clone(), actor.clone(), ); self.coordinator.persist(&mut self.store); let project = self .coordinator .project(&project) .expect("project was just created") .clone(); Ok(CoordinatorResponse::ProjectCreated { project, actor }) } CoordinatorRequest::SelectProject { tenant, actor_user, project, } => { let tenant = TenantId::new(tenant); let actor = UserId::new(actor_user); let project_id = ProjectId::new(project); let project = self .coordinator .project(&project_id) .ok_or_else(|| { CoordinatorError::Unauthorized( "project is not visible to the signed-in user".to_owned(), ) })? .clone(); if project.tenant != tenant { return Err(CoordinatorError::Unauthorized( "project is outside the signed-in tenant scope".to_owned(), ) .into()); } self.coordinator .upsert_user(tenant, actor.clone(), CredentialKind::BrowserSession); self.coordinator.persist(&mut self.store); Ok(CoordinatorResponse::ProjectSelected { project, actor }) } CoordinatorRequest::ListProjects { tenant, actor_user } => { let tenant = TenantId::new(tenant); let actor = UserId::new(actor_user); let context = disasmer_core::AuthContext { tenant: tenant.clone(), project: ProjectId::from("__project_listing__"), actor: Actor::User(actor.clone()), }; self.coordinator .upsert_user(tenant, actor.clone(), CredentialKind::BrowserSession); self.coordinator.persist(&mut self.store); Ok(CoordinatorResponse::Projects { projects: self.coordinator.list_projects(&context), actor, }) } CoordinatorRequest::AttachNode { tenant, project, node, public_key, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let node = NodeId::new(node); self.coordinator.upsert_tenant(tenant.clone()); self.coordinator.upsert_user( tenant.clone(), UserId::from("local-user"), CredentialKind::CliDeviceSession, ); self.coordinator .upsert_project(tenant.clone(), project.clone(), "local"); self.coordinator.upsert_source_provider_config( tenant.clone(), project.clone(), SourceProviderKind::Filesystem, Digest::sha256("local-filesystem"), ); self.coordinator.enroll_node( tenant.clone(), project.clone(), node.clone(), public_key, "node:attach", ); self.coordinator.persist(&mut self.store); Ok(CoordinatorResponse::NodeAttached { node, tenant, project, }) } CoordinatorRequest::CreateNodeEnrollmentGrant { tenant, project, actor_user, grant, now_epoch_seconds, ttl_seconds, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let actor = UserId::new(actor_user); self.coordinator.upsert_tenant(tenant.clone()); self.coordinator.upsert_user( tenant.clone(), actor, CredentialKind::CliDeviceSession, ); self.coordinator .upsert_project(tenant.clone(), project.clone(), "local"); self.coordinator.upsert_source_provider_config( tenant.clone(), project.clone(), SourceProviderKind::Filesystem, Digest::sha256("local-filesystem"), ); let scope = "node:attach".to_owned(); let expires_at_epoch_seconds = now_epoch_seconds.saturating_add(ttl_seconds); let enrollment = self.coordinator.create_node_enrollment_grant( tenant.clone(), project.clone(), grant.clone(), scope.clone(), expires_at_epoch_seconds, ); self.enrollment_grants .insert(enrollment_grant_key(&tenant, &project, &grant), enrollment); self.coordinator.persist(&mut self.store); Ok(CoordinatorResponse::NodeEnrollmentGrantCreated { tenant, project, grant, scope, expires_at_epoch_seconds, }) } CoordinatorRequest::ExchangeNodeEnrollmentGrant { tenant, project, node, public_key, enrollment_grant, now_epoch_seconds, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let node = NodeId::new(node); let grant = self .enrollment_grants .get_mut(&enrollment_grant_key(&tenant, &project, &enrollment_grant)) .ok_or(CoordinatorError::Enrollment( disasmer_core::EnrollmentError::Expired, ))?; let credential = self.coordinator.exchange_node_enrollment_grant( grant, node.clone(), &public_key, "node:attach", now_epoch_seconds, )?; self.coordinator.persist(&mut self.store); Ok(CoordinatorResponse::NodeEnrollmentExchanged { node, tenant, project, credential, }) } CoordinatorRequest::NodeHeartbeat { node } => { let node = NodeId::new(node); self.coordinator .node_identity(&node) .ok_or(CoordinatorError::UnknownNode)?; Ok(CoordinatorResponse::NodeHeartbeat { node, epoch: self.coordinator.coordinator_epoch(), }) } CoordinatorRequest::ReportNodeCapabilities { tenant, project, node, capabilities, cached_environment_digests, dependency_cache_digests, source_snapshots, artifact_locations, direct_connectivity, online, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let node = NodeId::new(node); let identity = self .coordinator .node_identity(&node) .ok_or(CoordinatorError::UnknownNode)?; if identity.tenant != tenant || identity.project != project { return Err(CoordinatorError::Unauthorized( "node capability report is outside the enrolled tenant/project scope" .to_owned(), ) .into()); } capabilities.validate_public_report()?; self.node_descriptors.insert( node.clone(), NodeDescriptor { id: node.clone(), tenant, project, capabilities, cached_environments: cached_environment_digests.into_iter().collect(), dependency_caches: dependency_cache_digests.into_iter().collect(), source_snapshots: source_snapshots.into_iter().collect(), artifact_locations: artifact_locations .into_iter() .map(ArtifactId::new) .collect(), direct_connectivity, online, }, ); Ok(CoordinatorResponse::NodeCapabilitiesRecorded { node, node_descriptors: self.node_descriptors.len(), }) } CoordinatorRequest::ListNodeDescriptors { tenant, project, actor_user, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let actor = UserId::new(actor_user); self.coordinator.upsert_user( tenant.clone(), actor.clone(), CredentialKind::BrowserSession, ); let descriptors = self .node_descriptors .values() .filter(|descriptor| { descriptor.tenant == tenant && descriptor.project == project }) .cloned() .collect(); Ok(CoordinatorResponse::NodeDescriptors { descriptors, actor }) } CoordinatorRequest::ScheduleTask { tenant, project, environment, environment_digest, required_capabilities, dependency_cache, source_snapshot, required_artifacts, quota_available, policy_allowed, prefer_node, } => { let request = PlacementRequest { tenant: TenantId::new(tenant), project: ProjectId::new(project), environment, environment_digest, required_capabilities: required_capabilities.into_iter().collect(), dependency_cache, source_snapshot, required_artifacts: required_artifacts .into_iter() .map(ArtifactId::new) .collect(), quota_available, policy_allowed, prefer_node: prefer_node.map(NodeId::new), }; let nodes = self.node_descriptors.values().cloned().collect::>(); let placement = DefaultScheduler.place(&nodes, &request)?; Ok(CoordinatorResponse::TaskPlacement { placement }) } CoordinatorRequest::LaunchTask { tenant, project, actor_user, process, task, environment, environment_digest, required_capabilities, dependency_cache, source_snapshot, required_artifacts, quota_available, policy_allowed, command, command_args, artifact_path, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let task = TaskId::new(task); let actor = UserId::new(actor_user); self.coordinator .upsert_user(tenant.clone(), actor, CredentialKind::BrowserSession); let active = self.coordinator.active_process(&process).ok_or_else(|| { CoordinatorError::Unauthorized( "task launch requires an active coordinator-side virtual process" .to_owned(), ) })?; if active.tenant != tenant || active.project != project { return Err(CoordinatorError::Unauthorized( "task launch is outside the virtual process tenant/project scope" .to_owned(), ) .into()); } let request = PlacementRequest { tenant: tenant.clone(), project: project.clone(), environment, environment_digest, required_capabilities: required_capabilities.into_iter().collect(), dependency_cache, source_snapshot, required_artifacts: required_artifacts .into_iter() .map(ArtifactId::new) .collect(), quota_available, policy_allowed, prefer_node: None, }; let nodes = self.node_descriptors.values().cloned().collect::>(); let placement = DefaultScheduler.place(&nodes, &request)?; let assignment = TaskAssignment { tenant: tenant.clone(), project: project.clone(), process: process.clone(), task: task.clone(), node: placement.node.clone(), epoch: active.coordinator_epoch, command, command_args, artifact_path, }; self.task_assignments .entry((tenant, project, placement.node.clone())) .or_default() .push_back(assignment.clone()); Ok(CoordinatorResponse::TaskLaunched { process, task, placement, assignment, }) } CoordinatorRequest::PollTaskAssignment { tenant, project, node, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let node = NodeId::new(node); let identity = self .coordinator .node_identity(&node) .ok_or(CoordinatorError::UnknownNode)?; if identity.tenant != tenant || identity.project != project { return Err(CoordinatorError::Unauthorized( "task assignment poll is outside the enrolled tenant/project scope" .to_owned(), ) .into()); } let assignment = self .task_assignments .get_mut(&(tenant, project, node)) .and_then(VecDeque::pop_front); Ok(CoordinatorResponse::TaskAssignment { assignment }) } CoordinatorRequest::RequestRendezvous { scope, source, destination, direct_connectivity, failure_reason, } => { self.rendezvous_meter.charge( &self.rendezvous_limits, LimitKind::RendezvousAttempt, 1, )?; let plan = self.transport.plan_authenticated_direct_bulk_transfer( RendezvousRequest { scope, source, destination, }, direct_connectivity, failure_reason, )?; Ok(CoordinatorResponse::RendezvousPlan { plan, charged_rendezvous_attempts: self .rendezvous_meter .used(&LimitKind::RendezvousAttempt), }) } CoordinatorRequest::RequestSourcePreparation { tenant, project, provider, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let preparation = SourcePreparation::node_task(tenant.clone(), project.clone(), provider); let request = PlacementRequest { tenant, project, environment: None, environment_digest: None, required_capabilities: preparation.required_capabilities.clone(), dependency_cache: None, source_snapshot: None, required_artifacts: Default::default(), quota_available: true, policy_allowed: true, prefer_node: None, }; let nodes = self.node_descriptors.values().cloned().collect::>(); let disposition = match DefaultScheduler.place(&nodes, &request) { Ok(placement) => SourcePreparationDisposition::Assigned { node: placement.node, }, Err(err) => SourcePreparationDisposition::Pending { reason: if err.message.is_empty() { "waiting for any capable node to prepare source".to_owned() } else { err.message }, }, }; Ok(CoordinatorResponse::SourcePreparation { status: SourcePreparationStatus { preparation, disposition, }, }) } CoordinatorRequest::CompleteSourcePreparation { tenant, project, node, provider, source_snapshot, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let node = NodeId::new(node); let identity = self .coordinator .node_identity(&node) .ok_or(CoordinatorError::UnknownNode)?; if identity.tenant != tenant || identity.project != project { return Err(CoordinatorError::Unauthorized( "source preparation completion is outside the enrolled tenant/project scope" .to_owned(), ) .into()); } let descriptor = self.node_descriptors.get_mut(&node).ok_or_else(|| { CoordinatorError::Unauthorized( "source preparation completion requires a node capability report" .to_owned(), ) })?; descriptor.source_snapshots.insert(source_snapshot.clone()); Ok(CoordinatorResponse::SourcePreparationCompleted { node, provider, source_snapshot, }) } CoordinatorRequest::StartProcess { tenant, project, process, } => { let process = ProcessId::new(process); self.coordinator.start_process( TenantId::new(tenant), ProjectId::new(project), process.clone(), ); Ok(CoordinatorResponse::ProcessStarted { process, epoch: self.coordinator.coordinator_epoch(), }) } CoordinatorRequest::ReconnectNode { node, process, epoch, } => { let node = NodeId::new(node); let process = ProcessId::new(process); self.coordinator .reconnect_node(&node, Some((&process, epoch)))?; Ok(CoordinatorResponse::NodeReconnected { node, process }) } CoordinatorRequest::CancelTask { tenant, project, process, node, task, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let node = NodeId::new(node); let task = TaskId::new(task); self.coordinator .authorize_node_for_process(&node, &tenant, &project, &process)?; let active = self.coordinator.active_process(&process).ok_or_else(|| { CoordinatorError::Unauthorized( "task cancellation requires an active virtual process".to_owned(), ) })?; if !active.connected_nodes.contains(&node) { return Err(CoordinatorError::Unauthorized( "task cancellation target node is not connected to the virtual process" .to_owned(), ) .into()); } self.task_cancellations .insert(task_control_key(&tenant, &project, &process, &node, &task)); Ok(CoordinatorResponse::TaskCancellationRequested { process, task, node, }) } CoordinatorRequest::PollTaskControl { tenant, project, process, node, task, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let node = NodeId::new(node); let task = TaskId::new(task); self.coordinator .authorize_node_for_process(&node, &tenant, &project, &process)?; let cancel_requested = self .task_cancellations .contains(&task_control_key(&tenant, &project, &process, &node, &task)); Ok(CoordinatorResponse::TaskControl { process, task, cancel_requested, }) } CoordinatorRequest::PollDebugCommand { tenant, project, process, node, task, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let node = NodeId::new(node); let task = TaskId::new(task); self.coordinator .authorize_node_for_process(&node, &tenant, &project, &process)?; Ok(CoordinatorResponse::DebugCommand { process, task, command: None, }) } CoordinatorRequest::ReportTaskLog { tenant, project, process, node, task, stdout_bytes, stderr_bytes, stdout_tail, stderr_tail, stdout_truncated, stderr_truncated, backpressured, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let node = NodeId::new(node); let task = TaskId::new(task); self.coordinator .authorize_node_for_process(&node, &tenant, &project, &process)?; validate_task_log_tail("stdout_tail", &stdout_tail)?; validate_task_log_tail("stderr_tail", &stderr_tail)?; Ok(CoordinatorResponse::TaskLogRecorded { process, task, stdout_bytes, stderr_bytes, stdout_tail: if stdout_truncated { format!("{stdout_tail}\n... truncated") } else { stdout_tail }, stderr_tail: if stderr_truncated { format!("{stderr_tail}\n... truncated") } else { stderr_tail }, backpressured, }) } CoordinatorRequest::ReportVfsMetadata { tenant, project, process, node, task, artifact_path, artifact_digest, artifact_size_bytes, large_bytes_uploaded, } => { let artifact_path = artifact_path .map(VfsPath::new) .transpose() .map_err(|err| CoordinatorServiceError::InvalidArtifactPath(err.to_string()))?; let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let node = NodeId::new(node); let task = TaskId::new(task); self.coordinator .authorize_node_for_process(&node, &tenant, &project, &process)?; if let (Some(path), Some(digest)) = (&artifact_path, artifact_digest) { self.artifact_registry.flush_metadata( artifact_id_from_path(path), tenant, project, process.clone(), task.clone(), node, digest, artifact_size_bytes.unwrap_or_default(), ); } Ok(CoordinatorResponse::VfsMetadataRecorded { process, task, artifact_path, large_bytes_uploaded, }) } CoordinatorRequest::TaskCompleted { tenant, project, process, node, task, terminal_state, status_code, stdout_bytes, stderr_bytes, stdout_tail, stderr_tail, stdout_truncated, stderr_truncated, artifact_path, artifact_digest, artifact_size_bytes, } => { validate_task_log_tail("stdout_tail", &stdout_tail)?; validate_task_log_tail("stderr_tail", &stderr_tail)?; let artifact_path = artifact_path .map(VfsPath::new) .transpose() .map_err(|err| CoordinatorServiceError::InvalidArtifactPath(err.to_string()))?; let event = TaskCompletionEvent { tenant: TenantId::new(tenant), project: ProjectId::new(project), process: ProcessId::new(process), node: NodeId::new(node), task: TaskId::new(task), terminal_state: terminal_state .unwrap_or_else(|| TaskTerminalState::from_status_code(status_code)), status_code, stdout_bytes, stderr_bytes, stdout_tail, stderr_tail, stdout_truncated, stderr_truncated, artifact_path, artifact_digest: artifact_digest.clone(), }; self.coordinator.authorize_node_for_process( &event.node, &event.tenant, &event.project, &event.process, )?; if let (Some(path), Some(digest)) = (&event.artifact_path, artifact_digest) { self.artifact_registry.flush_metadata( artifact_id_from_path(path), event.tenant.clone(), event.project.clone(), event.process.clone(), event.task.clone(), event.node.clone(), digest, artifact_size_bytes.unwrap_or(stdout_bytes), ); } self.task_cancellations.remove(&task_control_key( &event.tenant, &event.project, &event.process, &event.node, &event.task, )); self.task_events.push(event.clone()); Ok(CoordinatorResponse::TaskRecorded { process: event.process, task: event.task, events_recorded: self.task_events.len(), }) } CoordinatorRequest::ListTaskEvents { tenant, project, actor_user, process, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let actor = UserId::new(actor_user); let process = process.map(ProcessId::new); self.coordinator .upsert_user(tenant.clone(), actor, CredentialKind::BrowserSession); let events = self .task_events .iter() .filter(|event| { event.tenant == tenant && event.project == project && process .as_ref() .map_or(true, |process| event.process == *process) }) .cloned() .collect(); Ok(CoordinatorResponse::TaskEvents { events }) } CoordinatorRequest::RenderOperatorPanel { tenant, project, process, actor_user, max_download_bytes, stopped, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let stop_key = panel_stop_key(&tenant, &project, &process); let panel = if stopped { self.ensure_operator_panel_scope(&tenant, &project, &process)?; self.stopped_panels.insert(stop_key); let stop_key = panel_stop_key(&tenant, &project, &process); let panel = match self.panel_snapshots.get(&stop_key).cloned() { Some(panel) => stopped_panel_snapshot(panel), None => self.render_operator_panel( tenant.clone(), project.clone(), process.clone(), UserId::new(actor_user), max_download_bytes, true, )?, }; self.panel_snapshots.insert(stop_key, panel.clone()); panel } else { self.stopped_panels.remove(&stop_key); let panel = self.render_operator_panel( tenant.clone(), project.clone(), process.clone(), UserId::new(actor_user), max_download_bytes, false, )?; self.panel_snapshots.insert(stop_key, panel.clone()); panel }; Ok(CoordinatorResponse::OperatorPanel { panel }) } CoordinatorRequest::SubmitPanelEvent { tenant, project, process, widget_id, kind, max_events, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); let stop_key = panel_stop_key(&tenant, &project, &process); let stopped = self.stopped_panels.contains(&stop_key); let panel = if stopped { match self.panel_snapshots.get(&stop_key).cloned() { Some(panel) => stopped_panel_snapshot(panel), None => { let panel = self.render_operator_panel( tenant.clone(), project.clone(), process.clone(), UserId::from("panel-user"), self.download_limits .limit(&LimitKind::ArtifactDownloadBytes), true, )?; self.panel_snapshots.insert(stop_key.clone(), panel.clone()); panel } } } else { let panel = self.render_operator_panel( tenant.clone(), project.clone(), process.clone(), UserId::from("panel-user"), self.download_limits .limit(&LimitKind::ArtifactDownloadBytes), false, )?; self.panel_snapshots.insert(stop_key, panel.clone()); panel }; let event = PanelEvent { tenant: tenant.clone(), project: project.clone(), process: process.clone(), widget_id: widget_id.clone(), kind, }; let limit = self .panel_event_limits .entry((tenant, project, process, widget_id)) .or_insert(RateLimit { max_events, used_events: 0, }); limit.max_events = max_events; panel.accept_event(&event, limit)?; Ok(CoordinatorResponse::PanelEventAccepted { used_events: limit.used_events, max_events: limit.max_events, }) } CoordinatorRequest::CreateArtifactDownloadLink { tenant, project, actor_user, artifact, max_bytes, token_nonce, now_epoch_seconds, ttl_seconds, } => { let context = user_context(tenant, project, actor_user); let artifact = ArtifactId::new(artifact); let policy = DownloadPolicy { max_bytes }; let action = self .artifact_registry .download_action(&context, &artifact, &policy)?; self.ensure_download_source_connectivity(&action.source)?; let downloadable_size = self .artifact_registry .downloadable_size(&context, &artifact, &policy)?; self.download_meter.can_charge( &self.download_limits, LimitKind::ArtifactDownloadBytes, downloadable_size, )?; let link = self.artifact_registry.create_download_link( &context, &artifact, &policy, &token_nonce, now_epoch_seconds, ttl_seconds, )?; Ok(CoordinatorResponse::ArtifactDownloadLink { link }) } CoordinatorRequest::OpenArtifactDownloadStream { tenant, project, actor_user, artifact, max_bytes, token_nonce, token_digest, now_epoch_seconds, chunk_bytes, } => { let _ = token_nonce; let context = user_context(tenant, project, actor_user); let mut stream = self.artifact_registry.open_download_stream( &context, &ArtifactId::new(artifact), &DownloadPolicy { max_bytes }, &token_digest, now_epoch_seconds, &self.download_limits, &mut self.download_meter, )?; self.ensure_download_source_connectivity(&stream.link.source)?; self.artifact_registry.stream_download_chunk( &mut stream, &self.download_limits, &mut self.download_meter, chunk_bytes, )?; let charged_download_bytes = self.download_meter.used(&LimitKind::ArtifactDownloadBytes); Ok(CoordinatorResponse::ArtifactDownloadStream { link: stream.link, streamed_bytes: stream.streamed_bytes, charged_download_bytes, }) } CoordinatorRequest::RevokeArtifactDownloadLink { tenant, project, actor_user, artifact, token_digest, } => { let context = user_context(tenant, project, actor_user); let link = self.artifact_registry.revoke_download_link( &context, &ArtifactId::new(artifact), &token_digest, )?; Ok(CoordinatorResponse::ArtifactDownloadLinkRevoked { link }) } CoordinatorRequest::ExportArtifactToNode { tenant, project, actor_user, artifact, receiver_node, direct_connectivity, failure_reason, } => { let context = user_context(tenant, project, actor_user); let artifact = ArtifactId::new(artifact); let receiver_node = NodeId::new(receiver_node); let action = self.artifact_registry.download_action( &context, &artifact, &DownloadPolicy { max_bytes: u64::MAX, }, )?; let StorageLocation::RetainedNode(source_node) = action.source else { return Err(disasmer_core::DownloadError::Unavailable.into()); }; let metadata = self .artifact_registry .metadata(&artifact) .ok_or(disasmer_core::DownloadError::NotFound)?; let source = self.export_endpoint(&source_node, &context.tenant, &context.project)?; let destination = self.export_endpoint(&receiver_node, &context.tenant, &context.project)?; let plan = self.transport.plan_authenticated_direct_bulk_transfer( RendezvousRequest { scope: DataPlaneScope { tenant: context.tenant.clone(), project: context.project.clone(), process: metadata.process.clone(), object: DataPlaneObject::Artifact(artifact.clone()), authorization_subject: format!( "artifact-export:{}-to-{}", source_node, receiver_node ), }, source, destination, }, direct_connectivity, failure_reason, )?; Ok(CoordinatorResponse::ArtifactExportPlan { plan, source_node, receiver_node, }) } } } fn export_endpoint( &self, node: &NodeId, tenant: &TenantId, project: &ProjectId, ) -> Result { let identity = self .coordinator .node_identity(node) .ok_or(CoordinatorError::UnknownNode)?; if &identity.tenant != tenant || &identity.project != project { return Err(CoordinatorError::Unauthorized( "artifact export node is outside the tenant/project scope".to_owned(), ) .into()); } let descriptor = self.node_descriptors.get(node).ok_or_else(|| { disasmer_core::DownloadError::DirectConnectivityUnavailable(format!( "node {node} has not reported export connectivity" )) })?; if descriptor.tenant != *tenant || descriptor.project != *project { return Err(CoordinatorError::Unauthorized( "artifact export node descriptor is outside the tenant/project scope".to_owned(), ) .into()); } if !descriptor.online { return Err( disasmer_core::DownloadError::DirectConnectivityUnavailable(format!( "node {node} is offline for artifact export" )) .into(), ); } if !descriptor.direct_connectivity { return Err( disasmer_core::DownloadError::DirectConnectivityUnavailable(format!( "direct connectivity unavailable to node {node} for artifact export" )) .into(), ); } Ok(NodeEndpoint { node: node.clone(), advertised_addr: format!("{node}.mesh.invalid:4433"), public_key_fingerprint: Digest::sha256(&identity.public_key), }) } fn ensure_download_source_connectivity( &self, source: &StorageLocation, ) -> Result<(), disasmer_core::DownloadError> { let StorageLocation::RetainedNode(node) = source else { return Ok(()); }; let Some(descriptor) = self.node_descriptors.get(node) else { return Ok(()); }; if !descriptor.online { return Err(disasmer_core::DownloadError::DirectConnectivityUnavailable( format!("retaining node {node} is offline for artifact download"), )); } if !descriptor.direct_connectivity { return Err(disasmer_core::DownloadError::DirectConnectivityUnavailable( format!("direct connectivity unavailable to retaining node {node} for artifact download"), )); } Ok(()) } fn render_operator_panel( &self, tenant: TenantId, project: ProjectId, process: ProcessId, actor_user: UserId, max_download_bytes: u64, stopped: bool, ) -> Result { self.ensure_operator_panel_scope(&tenant, &project, &process)?; let events = self .task_events .iter() .filter(|event| { event.tenant == tenant && event.project == project && event.process == process }) .collect::>(); let completed = events .iter() .filter(|event| event.status_code == Some(0)) .count() as u64; let total = events.len().max(1) as u64; let stdout_bytes = events.iter().map(|event| event.stdout_bytes).sum::(); let stderr_bytes = events.iter().map(|event| event.stderr_bytes).sum::(); let last_task = events.last().map(|event| event.task.clone()); let mut panel = PanelState::new(tenant.clone(), project.clone(), process.clone()); if stopped { panel.freeze_program_ui_events(); } panel.add_widget(PanelWidget { id: "process-status".to_owned(), label: "Process Status".to_owned(), kind: PanelWidgetKind::Text { value: if stopped { "stopped".to_owned() } else { "running".to_owned() }, }, })?; panel.add_widget(PanelWidget { id: "task-progress".to_owned(), label: "Tasks".to_owned(), kind: PanelWidgetKind::Progress { current: completed, total, }, })?; panel.add_widget(PanelWidget { id: "task-summary".to_owned(), label: "Task Summary".to_owned(), kind: PanelWidgetKind::Text { value: if events.is_empty() { "no task events recorded".to_owned() } else { events .iter() .map(|event| { format!("{}:{:?}:{}", event.task, event.status_code, event.node) }) .collect::>() .join(", ") }, }, })?; panel.add_widget(PanelWidget { id: "recent-logs".to_owned(), label: "Recent Logs".to_owned(), kind: PanelWidgetKind::Text { value: format!("stdout={stdout_bytes} stderr={stderr_bytes}"), }, })?; panel.add_widget(PanelWidget { id: "debug-process".to_owned(), label: "Debug Process".to_owned(), kind: PanelWidgetKind::Button { action: "debug-process".to_owned(), }, })?; panel.add_widget(PanelWidget { id: "cancel-process".to_owned(), label: "Cancel Process".to_owned(), kind: PanelWidgetKind::Button { action: "cancel-process".to_owned(), }, })?; if last_task.is_some() { panel.add_widget(PanelWidget { id: "restart-selected-task".to_owned(), label: "Restart Selected Task".to_owned(), kind: PanelWidgetKind::Button { action: "restart-task".to_owned(), }, })?; } let mut actions = vec![ disasmer_core::ControlPlaneAction::DebugProcess, disasmer_core::ControlPlaneAction::CancelProcess, ]; if let Some(task) = last_task.clone() { actions.push(disasmer_core::ControlPlaneAction::RestartTask(task)); } panel.set_control_plane_actions(actions); if let Some(path) = events .iter() .rev() .find_map(|event| event.artifact_path.as_ref()) { let artifact = artifact_id_from_path(path); let context = disasmer_core::AuthContext { tenant, project, actor: Actor::User(actor_user), }; panel.add_download_widget_from_action( "download-artifact", "Download Artifact", self.artifact_registry.download_action( &context, &artifact, &DownloadPolicy { max_bytes: max_download_bytes, }, ), )?; } Ok(panel) } fn ensure_operator_panel_scope( &self, tenant: &TenantId, project: &ProjectId, process: &ProcessId, ) -> Result<(), CoordinatorServiceError> { let active = self.coordinator.active_process(process).ok_or_else(|| { CoordinatorError::Unauthorized( "operator panel requires an active virtual process".to_owned(), ) })?; if active.tenant != *tenant || active.project != *project { return Err(CoordinatorError::Unauthorized( "operator panel scope does not match the active virtual process".to_owned(), ) .into()); } Ok(()) } pub fn serve_tcp(self, listener: TcpListener) -> Result<(), CoordinatorServiceError> { let shared = Arc::new(Mutex::new(self)); for stream in listener.incoming() { let stream = stream?; let service = Arc::clone(&shared); std::thread::spawn(move || { if let Err(err) = handle_shared_stream(service, stream) { eprintln!("coordinator stream failed: {err}"); } }); } Ok(()) } pub fn handle_stream(&mut self, stream: TcpStream) -> Result<(), CoordinatorServiceError> { let mut reader = BufReader::new(stream.try_clone()?); let mut writer = stream; loop { let mut line = String::new(); if reader.read_line(&mut line)? == 0 { return Ok(()); } if line.trim().is_empty() { continue; } let response = match serde_json::from_str::(&line) { Ok(request) => match self.handle_request(request) { Ok(response) => response, Err(err) => CoordinatorResponse::Error { message: err.to_string(), }, }, Err(err) => CoordinatorResponse::Error { message: err.to_string(), }, }; serde_json::to_writer(&mut writer, &response)?; writer.write_all(b"\n")?; writer.flush()?; } } } fn handle_shared_stream( service: Arc>, stream: TcpStream, ) -> Result<(), CoordinatorServiceError> { let mut reader = BufReader::new(stream.try_clone()?); let mut writer = stream; loop { let mut line = String::new(); if reader.read_line(&mut line)? == 0 { return Ok(()); } if line.trim().is_empty() { continue; } let response = match serde_json::from_str::(&line) { Ok(request) => match service.lock() { Ok(mut service) => match service.handle_request(request) { Ok(response) => response, Err(err) => CoordinatorResponse::Error { message: err.to_string(), }, }, Err(_) => CoordinatorResponse::Error { message: "coordinator service lock poisoned".to_owned(), }, }, Err(err) => CoordinatorResponse::Error { message: err.to_string(), }, }; serde_json::to_writer(&mut writer, &response)?; writer.write_all(b"\n")?; writer.flush()?; } } pub fn bind_listener(addr: &str) -> Result<(TcpListener, SocketAddr), CoordinatorServiceError> { let listener = TcpListener::bind(addr)?; let addr = listener.local_addr()?; Ok((listener, addr)) } fn user_context(tenant: String, project: String, actor_user: String) -> disasmer_core::AuthContext { disasmer_core::AuthContext { tenant: TenantId::new(tenant), project: ProjectId::new(project), actor: Actor::User(UserId::new(actor_user)), } } fn artifact_id_from_path(path: &VfsPath) -> ArtifactId { let value = path .as_str() .strip_prefix("/vfs/artifacts/") .unwrap_or(path.as_str()) .replace('/', ":"); ArtifactId::new(value) } fn validate_task_log_tail(kind: &str, value: &str) -> Result<(), CoordinatorServiceError> { if value.len() > MAX_TASK_LOG_TAIL_BYTES { return Err(CoordinatorServiceError::InvalidTaskLogTail(format!( "{kind} is {} bytes; max is {MAX_TASK_LOG_TAIL_BYTES}", value.len() ))); } Ok(()) } fn task_control_key( tenant: &TenantId, project: &ProjectId, process: &ProcessId, node: &NodeId, task: &TaskId, ) -> TaskControlKey { ( tenant.clone(), project.clone(), process.clone(), node.clone(), task.clone(), ) } fn panel_stop_key(tenant: &TenantId, project: &ProjectId, process: &ProcessId) -> PanelStopKey { (tenant.clone(), project.clone(), process.clone()) } fn enrollment_grant_key(tenant: &TenantId, project: &ProjectId, grant: &str) -> EnrollmentGrantKey { (tenant.clone(), project.clone(), grant.to_owned()) } fn stopped_panel_snapshot(mut panel: PanelState) -> PanelState { panel.freeze_program_ui_events(); if let Some(status) = panel.widgets.get_mut("process-status") { status.kind = PanelWidgetKind::Text { value: "stopped".to_owned(), }; } panel } #[cfg(test)] mod tests { use std::collections::{BTreeMap, BTreeSet}; use disasmer_core::{DataPlaneObject, EnvironmentBackend, Os}; use super::*; fn linux_capabilities() -> NodeCapabilities { NodeCapabilities { os: Os::Linux, arch: "x86_64".to_owned(), capabilities: BTreeSet::from([ Capability::Command, Capability::Containers, Capability::RootlessPodman, Capability::VfsArtifacts, ]), environment_backends: BTreeSet::from([EnvironmentBackend::Container]), source_providers: BTreeSet::from(["filesystem".to_owned()]), } } fn windows_capabilities() -> NodeCapabilities { NodeCapabilities { os: Os::Windows, arch: "x86_64".to_owned(), capabilities: BTreeSet::from([ Capability::Command, Capability::WindowsCommandDev, Capability::VfsArtifacts, ]), environment_backends: BTreeSet::from([EnvironmentBackend::WindowsCommandDev]), source_providers: BTreeSet::from(["filesystem".to_owned()]), } } fn endpoint(name: &str) -> NodeEndpoint { NodeEndpoint { node: NodeId::from(name), advertised_addr: format!("{name}.mesh.invalid:4433"), public_key_fingerprint: Digest::sha256(format!("{name}-public-key")), } } fn data_plane_scope(project: &str) -> DataPlaneScope { DataPlaneScope { tenant: TenantId::from("tenant"), project: ProjectId::from(project), process: ProcessId::from("process"), object: DataPlaneObject::Artifact(ArtifactId::from("artifact")), authorization_subject: "node-a-to-node-b".to_owned(), } } #[test] fn service_creates_selects_and_lists_signed_in_user_projects() { let mut service = CoordinatorService::new(7); let CoordinatorResponse::ProjectCreated { project, actor } = service .handle_request(CoordinatorRequest::CreateProject { tenant: "tenant-a".to_owned(), actor_user: "user-a".to_owned(), project: "project-a".to_owned(), name: "Demo".to_owned(), }) .unwrap() else { panic!("expected project creation"); }; assert_eq!(actor, UserId::from("user-a")); assert_eq!(project.tenant, TenantId::from("tenant-a")); assert_eq!(project.id, ProjectId::from("project-a")); assert_eq!(project.name, "Demo"); let CoordinatorResponse::ProjectSelected { project, actor } = service .handle_request(CoordinatorRequest::SelectProject { tenant: "tenant-a".to_owned(), actor_user: "user-a".to_owned(), project: "project-a".to_owned(), }) .unwrap() else { panic!("expected project selection"); }; assert_eq!(actor, UserId::from("user-a")); assert_eq!(project.id, ProjectId::from("project-a")); service .handle_request(CoordinatorRequest::CreateProject { tenant: "tenant-b".to_owned(), actor_user: "user-b".to_owned(), project: "project-b".to_owned(), name: "Other".to_owned(), }) .unwrap(); let CoordinatorResponse::Projects { projects, actor } = service .handle_request(CoordinatorRequest::ListProjects { tenant: "tenant-a".to_owned(), actor_user: "user-a".to_owned(), }) .unwrap() else { panic!("expected project list"); }; assert_eq!(actor, UserId::from("user-a")); assert_eq!(projects.len(), 1); assert_eq!(projects[0].id, ProjectId::from("project-a")); let cross_tenant = service .handle_request(CoordinatorRequest::SelectProject { tenant: "tenant-a".to_owned(), actor_user: "user-a".to_owned(), project: "project-b".to_owned(), }) .unwrap_err(); assert!(cross_tenant.to_string().contains("tenant scope")); } #[test] fn service_attaches_node_starts_process_and_records_scoped_task_event() { let mut service = CoordinatorService::new(7); let attached = service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); assert!(matches!(attached, CoordinatorResponse::NodeAttached { .. })); let started = service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), }) .unwrap(); assert_eq!( started, CoordinatorResponse::ProcessStarted { process: ProcessId::from("process"), epoch: 7 } ); let heartbeat = service .handle_request(CoordinatorRequest::NodeHeartbeat { node: "node".to_owned(), }) .unwrap(); assert_eq!( heartbeat, CoordinatorResponse::NodeHeartbeat { node: NodeId::from("node"), epoch: 7, } ); service .handle_request(CoordinatorRequest::ReconnectNode { node: "node".to_owned(), process: "process".to_owned(), epoch: 7, }) .unwrap(); let recorded = service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: 12, stderr_bytes: 0, stdout_tail: "build ok".to_owned(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: Some("/vfs/artifacts/app.txt".to_owned()), artifact_digest: Some(Digest::sha256("artifact")), artifact_size_bytes: Some(12), }) .unwrap(); assert_eq!( recorded, CoordinatorResponse::TaskRecorded { process: ProcessId::from("process"), task: TaskId::from("compile-linux"), events_recorded: 1 } ); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: Some("process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert_eq!(events.len(), 1); assert_eq!(events[0].node, NodeId::from("node")); assert_eq!(events[0].stdout_tail, "build ok"); assert_eq!(events[0].stderr_tail, ""); assert!(!events[0].stdout_truncated); let oversized_tail = "x".repeat(MAX_TASK_LOG_TAIL_BYTES + 1); let oversized_log = service .handle_request(CoordinatorRequest::ReportTaskLog { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), stdout_bytes: oversized_tail.len() as u64, stderr_bytes: 0, stdout_tail: oversized_tail.clone(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, backpressured: false, }) .unwrap_err(); assert!(oversized_log.to_string().contains("stdout_tail")); let oversized_completion = service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: oversized_tail.len() as u64, stderr_bytes: 0, stdout_tail: oversized_tail, stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: None, artifact_digest: None, artifact_size_bytes: None, }) .unwrap_err(); assert!(oversized_completion.to_string().contains("stdout_tail")); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "other".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: Some("process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert!(events.is_empty()); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: Some("other-process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert!(events.is_empty()); } #[test] fn service_delivers_cancellation_to_connected_node_and_records_terminal_state() { let mut service = CoordinatorService::new(7); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::ReconnectNode { node: "node".to_owned(), process: "process".to_owned(), epoch: 7, }) .unwrap(); let cancelled = service .handle_request(CoordinatorRequest::CancelTask { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), }) .unwrap(); assert_eq!( cancelled, CoordinatorResponse::TaskCancellationRequested { process: ProcessId::from("process"), task: TaskId::from("compile-linux"), node: NodeId::from("node"), } ); let control = service .handle_request(CoordinatorRequest::PollTaskControl { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), }) .unwrap(); assert_eq!( control, CoordinatorResponse::TaskControl { process: ProcessId::from("process"), task: TaskId::from("compile-linux"), cancel_requested: true, } ); service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: Some(TaskTerminalState::Cancelled), status_code: None, stdout_bytes: 0, stderr_bytes: 0, stdout_tail: String::new(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: None, artifact_digest: None, artifact_size_bytes: None, }) .unwrap(); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: Some("process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert_eq!(events[0].terminal_state, TaskTerminalState::Cancelled); let control = service .handle_request(CoordinatorRequest::PollTaskControl { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), }) .unwrap(); assert_eq!( control, CoordinatorResponse::TaskControl { process: ProcessId::from("process"), task: TaskId::from("compile-linux"), cancel_requested: false, } ); } #[test] fn service_download_links_are_scoped_and_streaming_is_metered() { let mut service = CoordinatorService::new(7); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: 32, stderr_bytes: 0, stdout_tail: "artifact bytes".to_owned(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: Some("/vfs/artifacts/app.txt".to_owned()), artifact_digest: Some(Digest::sha256("artifact")), artifact_size_bytes: Some(32), }) .unwrap(); service.download_limits = ResourceLimits { limits: BTreeMap::from([(LimitKind::ArtifactDownloadBytes, 31)]), }; let quota_denied_link = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "quota-denied".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap_err(); assert!(quota_denied_link .to_string() .contains("ArtifactDownloadBytes")); service.download_limits = ResourceLimits { limits: BTreeMap::from([(LimitKind::ArtifactDownloadBytes, 64)]), }; service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: false, online: true, }) .unwrap(); let disconnected_link = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "disconnected".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap_err(); assert!(disconnected_link .to_string() .contains("direct connectivity unavailable")); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap(); let CoordinatorResponse::ArtifactDownloadLink { link } = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap() else { panic!("expected download link"); }; service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: false, online: true, }) .unwrap(); let disconnected_open = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 1, }) .unwrap_err(); assert!(disconnected_open .to_string() .contains("direct connectivity unavailable")); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap(); assert_eq!(link.tenant, TenantId::from("tenant")); assert_eq!(link.project, ProjectId::from("project")); assert_eq!(link.process, ProcessId::from("process")); assert_eq!(link.actor, Actor::User(UserId::from("user"))); assert_eq!(link.max_bytes, 64); assert!(link.policy_context_digest.is_valid_sha256()); assert_eq!(link.expires_at_epoch_seconds, 70); assert!(link.url_path.contains("/artifacts/tenant/project/process")); let cross_tenant = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "other".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap_err(); assert!(cross_tenant.to_string().contains("tenant mismatch")); let cross_project = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "other-project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap_err(); assert!(cross_project.to_string().contains("project mismatch")); let cross_tenant_open = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "other".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 1, }) .unwrap_err(); assert!(cross_tenant_open.to_string().contains("tenant mismatch")); let cross_project_open = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "other-project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 1, }) .unwrap_err(); assert!(cross_project_open.to_string().contains("project mismatch")); let guessed = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: Digest::sha256("guessed"), now_epoch_seconds: 11, chunk_bytes: 1, }) .unwrap_err(); assert!(guessed.to_string().contains("token is invalid")); let cross_actor_open = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "other-user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 1, }) .unwrap_err(); assert!(cross_actor_open.to_string().contains("token is invalid")); let expired = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 71, chunk_bytes: 1, }) .unwrap_err(); assert!(expired.to_string().contains("expired")); let CoordinatorResponse::ArtifactDownloadStream { streamed_bytes, charged_download_bytes, .. } = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 16, }) .unwrap() else { panic!("expected download stream"); }; assert_eq!(streamed_bytes, 16); assert_eq!(charged_download_bytes, 16); let cross_actor_revoke = service .handle_request(CoordinatorRequest::RevokeArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "other-user".to_owned(), artifact: "app.txt".to_owned(), token_digest: link.scoped_token_digest.clone(), }) .unwrap_err(); assert!(cross_actor_revoke.to_string().contains("token is invalid")); let CoordinatorResponse::ArtifactDownloadLinkRevoked { link: revoked } = service .handle_request(CoordinatorRequest::RevokeArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), token_digest: link.scoped_token_digest.clone(), }) .unwrap() else { panic!("expected revoked download link"); }; assert_eq!(revoked.scoped_token_digest, link.scoped_token_digest); let revoked = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 64, token_nonce: "nonce".to_owned(), token_digest: link.scoped_token_digest, now_epoch_seconds: 12, chunk_bytes: 1, }) .unwrap_err(); assert!(revoked.to_string().contains("revoked")); } #[test] fn download_quota_cannot_be_reset_by_reconnect_retry_or_scope_switch() { let mut service = CoordinatorService::new(7); service.download_limits = ResourceLimits { limits: BTreeMap::from([(LimitKind::ArtifactDownloadBytes, 16)]), }; service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap(); service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: 16, stderr_bytes: 0, stdout_tail: "artifact bytes".to_owned(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: Some("/vfs/artifacts/app.txt".to_owned()), artifact_digest: Some(Digest::sha256("artifact")), artifact_size_bytes: Some(16), }) .unwrap(); let CoordinatorResponse::ArtifactDownloadLink { link } = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "first-cli-session".to_owned(), now_epoch_seconds: 10, ttl_seconds: 60, }) .unwrap() else { panic!("expected initial download link"); }; let CoordinatorResponse::ArtifactDownloadStream { charged_download_bytes, .. } = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "first-cli-session".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 11, chunk_bytes: 16, }) .unwrap() else { panic!("expected initial download stream"); }; assert_eq!(charged_download_bytes, 16); service .handle_request(CoordinatorRequest::ReconnectNode { node: "node".to_owned(), process: "process".to_owned(), epoch: 7, }) .unwrap(); let after_reconnect = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "after-reconnect".to_owned(), now_epoch_seconds: 12, ttl_seconds: 60, }) .unwrap_err(); assert!(after_reconnect .to_string() .contains("ArtifactDownloadBytes")); let restarted_cli_retry = service .handle_request(CoordinatorRequest::CreateArtifactDownloadLink { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "second-cli-session".to_owned(), now_epoch_seconds: 13, ttl_seconds: 60, }) .unwrap_err(); assert!(restarted_cli_retry .to_string() .contains("ArtifactDownloadBytes")); let cross_tenant_retry = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "other-tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "first-cli-session".to_owned(), token_digest: link.scoped_token_digest.clone(), now_epoch_seconds: 14, chunk_bytes: 1, }) .unwrap_err(); assert!(cross_tenant_retry.to_string().contains("tenant mismatch")); let cross_project_retry = service .handle_request(CoordinatorRequest::OpenArtifactDownloadStream { tenant: "tenant".to_owned(), project: "other-project".to_owned(), actor_user: "user".to_owned(), artifact: "app.txt".to_owned(), max_bytes: 16, token_nonce: "first-cli-session".to_owned(), token_digest: link.scoped_token_digest, now_epoch_seconds: 14, chunk_bytes: 1, }) .unwrap_err(); assert!(cross_project_retry.to_string().contains("project mismatch")); } #[test] fn windows_task_events_share_the_virtual_process_scope() { let mut service = CoordinatorService::new(7); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "windows-node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "windows-node".to_owned(), capabilities: windows_capabilities(), cached_environment_digests: vec![Digest::sha256("env-windows-command-dev")], dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), }) .unwrap(); let recorded = service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), node: "windows-node".to_owned(), task: "windows-command-dev".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: 24, stderr_bytes: 0, stdout_tail: "windows output".to_owned(), stderr_tail: String::new(), stdout_truncated: false, stderr_truncated: false, artifact_path: Some("/vfs/artifacts/windows-output.txt".to_owned()), artifact_digest: Some(Digest::sha256("windows-artifact")), artifact_size_bytes: Some(24), }) .unwrap(); assert_eq!( recorded, CoordinatorResponse::TaskRecorded { process: ProcessId::from("process"), task: TaskId::from("windows-command-dev"), events_recorded: 1, } ); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: Some("process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert_eq!(events.len(), 1); assert_eq!(events[0].process, ProcessId::from("process")); assert_eq!(events[0].node, NodeId::from("windows-node")); assert_eq!(events[0].task, TaskId::from("windows-command-dev")); assert_eq!( events[0].artifact_path, Some(VfsPath::new("/vfs/artifacts/windows-output.txt").unwrap()) ); } #[test] fn service_schedules_task_across_reported_node_descriptors() { let mut service = CoordinatorService::new(7); for node in ["cold-node", "warm-node"] { service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: node.to_owned(), public_key: format!("{node}-public-key"), }) .unwrap(); } let recorded = service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "cold-node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: false, online: true, }) .unwrap(); assert_eq!( recorded, CoordinatorResponse::NodeCapabilitiesRecorded { node: NodeId::from("cold-node"), node_descriptors: 1, } ); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "warm-node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: vec![Digest::sha256("env")], dependency_cache_digests: vec![Digest::sha256("deps")], source_snapshots: vec![Digest::sha256("source")], artifact_locations: vec!["build-output".to_owned()], direct_connectivity: true, online: true, }) .unwrap(); let CoordinatorResponse::NodeDescriptors { descriptors, actor } = service .handle_request(CoordinatorRequest::ListNodeDescriptors { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "operator".to_owned(), }) .unwrap() else { panic!("expected node descriptor inspector state"); }; assert_eq!(actor, UserId::from("operator")); assert_eq!(descriptors.len(), 2); let warm = descriptors .iter() .find(|descriptor| descriptor.id == NodeId::from("warm-node")) .expect("warm node descriptor is visible to inspector state"); assert!(warm .capabilities .capabilities .contains(&Capability::Command)); assert!(warm.cached_environments.contains(&Digest::sha256("env"))); assert!(warm.dependency_caches.contains(&Digest::sha256("deps"))); assert!(warm.source_snapshots.contains(&Digest::sha256("source"))); assert!(warm .artifact_locations .contains(&ArtifactId::from("build-output"))); let CoordinatorResponse::TaskPlacement { placement } = service .handle_request(CoordinatorRequest::ScheduleTask { tenant: "tenant".to_owned(), project: "project".to_owned(), environment: Some(EnvironmentRequirements::linux_container()), environment_digest: Some(Digest::sha256("env")), required_capabilities: vec![Capability::Command], dependency_cache: Some(Digest::sha256("deps")), source_snapshot: Some(Digest::sha256("source")), required_artifacts: vec!["build-output".to_owned()], quota_available: true, policy_allowed: true, prefer_node: None, }) .unwrap() else { panic!("expected task placement"); }; assert_eq!(placement.node, NodeId::from("warm-node")); assert!(placement .reasons .iter() .any(|reason| reason.contains("environment"))); assert!(placement .reasons .iter() .any(|reason| reason.contains("source"))); assert!(placement .reasons .iter() .any(|reason| reason.contains("dependency"))); assert!(placement .reasons .iter() .any(|reason| reason.contains("artifact"))); let error = service .handle_request(CoordinatorRequest::ScheduleTask { tenant: "tenant".to_owned(), project: "project".to_owned(), environment: None, environment_digest: None, required_capabilities: vec![Capability::WindowsCommandDev], dependency_cache: None, source_snapshot: None, required_artifacts: Vec::new(), quota_available: true, policy_allowed: true, prefer_node: None, }) .unwrap_err(); assert!(error.to_string().contains("WindowsCommandDev")); let quota_error = service .handle_request(CoordinatorRequest::ScheduleTask { tenant: "tenant".to_owned(), project: "project".to_owned(), environment: None, environment_digest: None, required_capabilities: vec![Capability::Command], dependency_cache: None, source_snapshot: None, required_artifacts: Vec::new(), quota_available: false, policy_allowed: true, prefer_node: None, }) .unwrap_err(); assert!(quota_error .to_string() .contains("quota unavailable for placement")); let policy_error = service .handle_request(CoordinatorRequest::ScheduleTask { tenant: "tenant".to_owned(), project: "project".to_owned(), environment: None, environment_digest: None, required_capabilities: vec![Capability::Command], dependency_cache: None, source_snapshot: None, required_artifacts: Vec::new(), quota_available: true, policy_allowed: false, prefer_node: None, }) .unwrap_err(); assert!(policy_error.to_string().contains("policy denied placement")); } #[test] fn coordinator_side_task_launch_queues_worker_assignment() { let mut service = CoordinatorService::new(9); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "worker-linux".to_owned(), public_key: "worker-linux-public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "worker-linux".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap(); let CoordinatorResponse::ProcessStarted { epoch, .. } = service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "vp-control".to_owned(), }) .unwrap() else { panic!("expected coordinator-side process start"); }; let CoordinatorResponse::TaskLaunched { process, task, placement, assignment, } = service .handle_request(CoordinatorRequest::LaunchTask { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: "vp-control".to_owned(), task: "compile-linux".to_owned(), environment: None, environment_digest: None, required_capabilities: vec![Capability::Command], dependency_cache: None, source_snapshot: None, required_artifacts: Vec::new(), quota_available: true, policy_allowed: true, command: "cargo".to_owned(), command_args: vec!["test".to_owned()], artifact_path: "/vfs/artifacts/dap-output.txt".to_owned(), }) .unwrap() else { panic!("expected launched task"); }; assert_eq!(process, ProcessId::from("vp-control")); assert_eq!(task, TaskId::from("compile-linux")); assert_eq!(placement.node, NodeId::from("worker-linux")); assert_eq!(assignment.node, NodeId::from("worker-linux")); assert_eq!(assignment.epoch, epoch); let CoordinatorResponse::TaskAssignment { assignment } = service .handle_request(CoordinatorRequest::PollTaskAssignment { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "worker-linux".to_owned(), }) .unwrap() else { panic!("expected worker assignment poll"); }; let assignment = assignment.expect("worker should receive queued assignment"); assert_eq!(assignment.process, ProcessId::from("vp-control")); assert_eq!(assignment.task, TaskId::from("compile-linux")); assert_eq!(assignment.command, "cargo"); assert_eq!(assignment.command_args, vec!["test"]); let CoordinatorResponse::TaskAssignment { assignment } = service .handle_request(CoordinatorRequest::PollTaskAssignment { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "worker-linux".to_owned(), }) .unwrap() else { panic!("expected empty worker assignment poll"); }; assert!(assignment.is_none()); } #[test] fn coordinator_side_task_launch_fails_cleanly_without_capable_worker() { let mut service = CoordinatorService::new(10); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "vp-control".to_owned(), }) .unwrap(); let error = service .handle_request(CoordinatorRequest::LaunchTask { tenant: "tenant".to_owned(), project: "project".to_owned(), actor_user: "user".to_owned(), process: "vp-control".to_owned(), task: "compile-linux".to_owned(), environment: None, environment_digest: None, required_capabilities: vec![Capability::Command], dependency_cache: None, source_snapshot: None, required_artifacts: Vec::new(), quota_available: true, policy_allowed: true, command: "cargo".to_owned(), command_args: vec!["test".to_owned()], artifact_path: "/vfs/artifacts/dap-output.txt".to_owned(), }) .unwrap_err(); assert!(error.to_string().contains("no capable node")); } #[test] fn service_rejects_malformed_node_capability_report() { let mut service = CoordinatorService::new(1); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); let mut capabilities = linux_capabilities(); capabilities .source_providers .insert("../checkout".to_owned()); let error = service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), capabilities, cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap_err(); assert!(error.to_string().contains("source provider id")); assert!(service.node_descriptors.is_empty()); } #[test] fn service_meters_rendezvous_before_direct_transfer_plan() { let mut service = CoordinatorService::new(7); service.rendezvous_limits = ResourceLimits { limits: BTreeMap::from([(LimitKind::RendezvousAttempt, 2)]), }; let CoordinatorResponse::RendezvousPlan { plan, charged_rendezvous_attempts, } = service .handle_request(CoordinatorRequest::RequestRendezvous { scope: data_plane_scope("project"), source: endpoint("node-a"), destination: endpoint("node-b"), direct_connectivity: true, failure_reason: String::new(), }) .unwrap() else { panic!("expected rendezvous plan"); }; assert_eq!(charged_rendezvous_attempts, 1); assert_eq!(plan.scope.project, ProjectId::from("project")); assert_eq!(plan.source.node, NodeId::from("node-a")); assert_eq!(plan.destination.node, NodeId::from("node-b")); assert!(plan.coordinator_assisted_rendezvous); assert!(!plan.coordinator_bulk_relay_allowed); let unavailable = service .handle_request(CoordinatorRequest::RequestRendezvous { scope: data_plane_scope("project"), source: endpoint("node-a"), destination: endpoint("node-b"), direct_connectivity: false, failure_reason: "nat traversal failed".to_owned(), }) .unwrap_err(); let unavailable = unavailable.to_string(); assert!(unavailable.contains("nat traversal failed")); assert!(unavailable.contains("coordinator bulk relay is disabled")); let quota = service .handle_request(CoordinatorRequest::RequestRendezvous { scope: data_plane_scope("project"), source: endpoint("node-a"), destination: endpoint("node-b"), direct_connectivity: true, failure_reason: String::new(), }) .unwrap_err(); assert!(quota.to_string().contains("RendezvousAttempt")); } #[test] fn service_rejects_task_completion_outside_node_scope() { let mut service = CoordinatorService::new(1); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant-a".to_owned(), project: "project-a".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); service .handle_request(CoordinatorRequest::StartProcess { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), process: "process".to_owned(), }) .unwrap(); let error = service .handle_request(CoordinatorRequest::TaskCompleted { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), process: "process".to_owned(), node: "node".to_owned(), task: "compile-linux".to_owned(), terminal_state: None, status_code: Some(0), stdout_bytes: 128, stderr_bytes: 64, stdout_tail: "foreign stdout".to_owned(), stderr_tail: "foreign stderr".to_owned(), stdout_truncated: false, stderr_truncated: false, artifact_path: Some("/vfs/artifacts/foreign.txt".to_owned()), artifact_digest: Some(Digest::sha256("foreign-artifact")), artifact_size_bytes: Some(128), }) .unwrap_err(); assert!(error.to_string().contains("outside")); let CoordinatorResponse::TaskEvents { events } = service .handle_request(CoordinatorRequest::ListTaskEvents { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), actor_user: "user".to_owned(), process: Some("process".to_owned()), }) .unwrap() else { panic!("expected task events"); }; assert!(events.is_empty()); } #[test] fn service_rejects_node_capability_report_outside_enrollment_scope() { let mut service = CoordinatorService::new(1); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant-a".to_owned(), project: "project-a".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); let error = service .handle_request(CoordinatorRequest::ReportNodeCapabilities { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), node: "node".to_owned(), capabilities: linux_capabilities(), cached_environment_digests: Vec::new(), dependency_cache_digests: Vec::new(), source_snapshots: Vec::new(), artifact_locations: Vec::new(), direct_connectivity: true, online: true, }) .unwrap_err(); assert!(error.to_string().contains("tenant/project scope")); assert!(service.node_descriptors.is_empty()); } #[test] fn service_rejects_source_preparation_completion_outside_node_scope() { let mut service = CoordinatorService::new(1); service .handle_request(CoordinatorRequest::AttachNode { tenant: "tenant-a".to_owned(), project: "project-a".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }) .unwrap(); let error = service .handle_request(CoordinatorRequest::CompleteSourcePreparation { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), node: "node".to_owned(), provider: SourceProviderKind::Filesystem, source_snapshot: Digest::sha256("foreign-source"), }) .unwrap_err(); assert!(error.to_string().contains("tenant/project scope")); } #[test] fn service_rejects_unknown_node_heartbeat() { let mut service = CoordinatorService::new(1); let error = service .handle_request(CoordinatorRequest::NodeHeartbeat { node: "missing".to_owned(), }) .unwrap_err(); assert!(error.to_string().contains("not enrolled")); } #[test] fn service_stream_accepts_multiple_requests_on_one_connection() { let (listener, addr) = bind_listener("127.0.0.1:0").unwrap(); let server = std::thread::spawn(move || { let mut service = CoordinatorService::new(3); let (stream, _) = listener.accept().unwrap(); service.handle_stream(stream).unwrap(); }); let mut stream = TcpStream::connect(addr).unwrap(); let mut reader = BufReader::new(stream.try_clone().unwrap()); serde_json::to_writer( &mut stream, &CoordinatorRequest::AttachNode { tenant: "tenant".to_owned(), project: "project".to_owned(), node: "node".to_owned(), public_key: "public-key".to_owned(), }, ) .unwrap(); stream.write_all(b"\n").unwrap(); stream.flush().unwrap(); let mut line = String::new(); reader.read_line(&mut line).unwrap(); assert!(matches!( serde_json::from_str::(&line).unwrap(), CoordinatorResponse::NodeAttached { .. } )); serde_json::to_writer( &mut stream, &CoordinatorRequest::NodeHeartbeat { node: "node".to_owned(), }, ) .unwrap(); stream.write_all(b"\n").unwrap(); stream.flush().unwrap(); line.clear(); reader.read_line(&mut line).unwrap(); assert_eq!( serde_json::from_str::(&line).unwrap(), CoordinatorResponse::NodeHeartbeat { node: NodeId::from("node"), epoch: 3, } ); stream.shutdown(std::net::Shutdown::Both).unwrap(); server.join().unwrap(); } }