diff --git a/Cargo.lock b/Cargo.lock index 3065c72..4bb96dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1384,9 +1384,9 @@ dependencies = [ [[package]] name = "futures" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" dependencies = [ "futures-channel", "futures-core", @@ -1412,9 +1412,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", "futures-sink", @@ -1422,15 +1422,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" [[package]] name = "futures-executor" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" dependencies = [ "futures-core", "futures-task", @@ -1439,9 +1439,9 @@ dependencies = [ [[package]] name = "futures-io" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" [[package]] name = "futures-lite" @@ -1458,9 +1458,9 @@ dependencies = [ [[package]] name = "futures-macro" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", @@ -1469,21 +1469,21 @@ dependencies = [ [[package]] name = "futures-sink" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" [[package]] name = "futures-task" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" [[package]] name = "futures-util" -version = "0.3.31" +version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-channel", "futures-core", @@ -1493,7 +1493,6 @@ dependencies = [ "futures-task", "memchr", "pin-project-lite", - "pin-utils", "slab", ] @@ -3787,6 +3786,7 @@ dependencies = [ "serde", "serde_json", "swactor", + "swactor-ci", "tiny_http", "tracing", "tracing-subscriber", @@ -4058,6 +4058,19 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "sha1_smol" version = "1.0.1" @@ -4187,6 +4200,7 @@ dependencies = [ "serde_json", "simulation", "swactor", + "swactor-ci", "tiny_http", "toml", ] @@ -4391,6 +4405,15 @@ dependencies = [ "ed25519-dalek 2.2.0", ] +name = "swactor-ci" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "serde_yaml", + "swactor", +] + [[package]] name = "swactor-datastore" version = "0.1.0" @@ -4992,6 +5015,12 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/Cargo.toml b/Cargo.toml index 589030e..e7a8439 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,20 @@ [workspace] -members = [".", "crates/python", "crates/wasm", "crates/bin-runner", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/std", "crates/datastore", "crates/shared-types", "crates/crypto-wasm", "tests/docker", "xtask"] +members = [ + ".", + "crates/python", + "crates/wasm", + "crates/bin-runner", + "crates/simulation", + "crates/runtime-dashboard", + "crates/distribution", + "crates/std", + "crates/datastore", + "crates/shared-types", + "crates/crypto-wasm", + "tests/docker", + "crates/ci", + "xtask", +] exclude = ["tools/depgraph"] [package] @@ -9,8 +24,8 @@ edition = "2024" autobenches = false [profile.bench] -debug = true -strip = false +debug = true +strip = false [lib] crate-type = ["rlib"] @@ -21,7 +36,8 @@ getrandom = ["dep:getrandom"] serde = ["dep:serde"] tracing = ["dep:tracing"] no_random = [] # compile without access to a source of randomness -transport = [] # transport-agnostic messaging (no mandatory deps; codec is user-provided) +transport = [ +] # transport-agnostic messaging (no mandatory deps; codec is user-provided) wasm = ["no_random", "dep:web-time"] # browser/wasm32 target support [dependencies] diff --git a/crates/ci/Cargo.toml b/crates/ci/Cargo.toml new file mode 100644 index 0000000..08550a6 --- /dev/null +++ b/crates/ci/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "swactor-ci" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../..", features = ["serde"] } +serde = { version = "1", features = ["derive"] } +serde_yaml = "0.9" +serde_json = "1" + +[dev-dependencies] diff --git a/crates/ci/src/coordinator.rs b/crates/ci/src/coordinator.rs new file mode 100644 index 0000000..1b63fd1 --- /dev/null +++ b/crates/ci/src/coordinator.rs @@ -0,0 +1,426 @@ +//! Coordinator actor: central brain of the CI system. +//! +//! Receives webhook events, manages pipeline lifecycles, dispatches jobs +//! to the Provisioner and RunnerSupervisor actors. + +use std::collections::HashMap; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use crate::pipeline::PipelineExecution; +use crate::yaml::{self, CiYaml}; +use crate::{ + CiConfig, JobComplete, JobId, JobProgress, JobStatus, PipelineId, ProvisionRequest, + ProvisionResponse, StatusUpdate, TerminateRequest, WebhookEvent, +}; + +/// Messages the Coordinator can receive. +#[derive(Debug, Clone)] +pub enum CoordinatorMsg { + /// A webhook event from Forgejo. + Webhook(WebhookEvent), + /// The CI YAML config to use (loaded externally or from the repo). + SetCiYaml(CiYaml), + /// Response from the Provisioner. + ProvisionResponse(ProvisionResponse), + /// Streamed output from a RunnerSupervisor. + JobProgress(JobProgress), + /// Final result from a RunnerSupervisor. + JobComplete(JobComplete), + /// Notification that the provisioner is offline (detected via SWIM). + ProvisionerOffline, + /// Notification that the provisioner is back online. + ProvisionerOnline, +} + +/// The Coordinator actor state. +pub struct Coordinator { + config: CiConfig, + ci_yaml: Option, + pipelines: HashMap, + next_pipeline_id: u64, + provisioner_addr: Option, + provisioner_online: bool, + /// Maps job_id → runner supervisor address. + runner_addrs: HashMap, + /// Captured status updates (for testing/simulation). + status_updates: Vec, + /// Jobs waiting for the provisioner to come online. + queued_provisions: Vec, +} + +impl Coordinator { + pub fn new(config: CiConfig) -> Self { + Self { + config, + ci_yaml: None, + pipelines: HashMap::new(), + next_pipeline_id: 1, + provisioner_addr: None, + provisioner_online: false, + runner_addrs: HashMap::new(), + status_updates: Vec::new(), + queued_provisions: Vec::new(), + } + } + + pub fn with_provisioner(mut self, addr: ActorAddress) -> Self { + self.provisioner_addr = Some(addr); + self.provisioner_online = true; + self + } + + pub fn with_ci_yaml(mut self, yaml: CiYaml) -> Self { + self.ci_yaml = Some(yaml); + self + } + + pub fn pipelines(&self) -> &HashMap { + &self.pipelines + } + + pub fn status_updates(&self) -> &[StatusUpdate] { + &self.status_updates + } + + fn handle_webhook(&mut self, ctx: &Ctx, event: WebhookEvent) { + let ci = match &self.ci_yaml { + Some(ci) => ci.clone(), + None => return, + }; + + let matched = yaml::matching_pipelines(&ci, &event); + for pipeline_name in matched { + let pipeline_def = &ci.pipelines[&pipeline_name]; + let pipeline_id = PipelineId(self.next_pipeline_id); + self.next_pipeline_id += 1; + + // Build job definitions. + let job_defs: Vec<_> = pipeline_def + .jobs + .iter() + .map(|(name, def)| yaml::to_job_definition(name, def)) + .collect(); + + let pipeline = PipelineExecution::new( + pipeline_id, + pipeline_name.clone(), + event.repo_owner.clone(), + event.repo_name.clone(), + event.commit_sha.clone(), + event.branch.clone(), + job_defs, + ); + + // Set pending status on Forgejo. + self.emit_status_update(StatusUpdate { + repo_owner: event.repo_owner.clone(), + repo_name: event.repo_name.clone(), + commit_sha: event.commit_sha.clone(), + state: "pending".into(), + context: format!("ci/{pipeline_name}"), + description: format!("Pipeline '{pipeline_name}' is pending"), + }); + + self.pipelines.insert(pipeline_id, pipeline); + + // Start eligible jobs. + self.advance_pipeline(ctx, pipeline_id); + } + } + + fn advance_pipeline(&mut self, ctx: &Ctx, pipeline_id: PipelineId) { + let pipeline = match self.pipelines.get(&pipeline_id) { + Some(p) => p, + None => return, + }; + + // If pipeline is already terminal, emit final status. + if pipeline.status.is_terminal() { + let update = StatusUpdate { + repo_owner: pipeline.repo_owner.clone(), + repo_name: pipeline.repo_name.clone(), + commit_sha: pipeline.commit_sha.clone(), + state: pipeline.status.forgejo_state().into(), + context: format!("ci/{}", pipeline.pipeline_name), + description: format!( + "Pipeline '{}' {}", + pipeline.pipeline_name, + pipeline.status.forgejo_state() + ), + }; + self.emit_status_update(update); + return; + } + + let eligible = pipeline.eligible_jobs(); + let repo_owner = pipeline.repo_owner.clone(); + let repo_name = pipeline.repo_name.clone(); + let commit_sha = pipeline.commit_sha.clone(); + + for job_name in eligible { + let job_id = JobId { + pipeline_id, + job_name: job_name.clone(), + }; + + let spec = { + let pipeline = self.pipelines.get(&pipeline_id).unwrap(); + let job = &pipeline.jobs[&job_name]; + crate::InstanceSpec { + docker_required: job.definition.docker, + ..Default::default() + } + }; + + if self.provisioner_online { + // Request provisioning. + if let Some(prov_addr) = self.provisioner_addr { + let request = ProvisionRequest { + job_id: job_id.clone(), + instance_spec: spec, + }; + let _ = ctx.send(prov_addr, crate::provisioner::ProvisionerMsg::Provision(request)); + } + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.status = JobStatus::Provisioning; + } + } + } else { + // Queue for later. + let request = ProvisionRequest { + job_id: job_id.clone(), + instance_spec: spec, + }; + self.queued_provisions.push(request); + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.status = JobStatus::WaitingForProvisioner; + } + } + } + + // Emit per-job status. + self.emit_status_update(StatusUpdate { + repo_owner: repo_owner.clone(), + repo_name: repo_name.clone(), + commit_sha: commit_sha.clone(), + state: "pending".into(), + context: format!("ci/{job_name}"), + description: format!("Job '{job_name}' is provisioning"), + }); + } + } + + fn handle_provision_response(&mut self, ctx: &Ctx, response: ProvisionResponse) { + let pipeline_id = response.job_id.pipeline_id; + let job_name = response.job_id.job_name.clone(); + + match response.result { + Ok(instance) => { + // Store instance_id for cleanup. + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.instance_id = Some(instance.instance_id.clone()); + job.status = JobStatus::Running; + } + } + + // Spawn a RunnerSupervisor for this job. + let pipeline = &self.pipelines[&pipeline_id]; + let job = &pipeline.jobs[&job_name]; + + let start_job = crate::StartJob { + job_id: response.job_id.clone(), + instance: instance.clone(), + repo_url: format!( + "{}/{}/{}", + self.config.forgejo_url, pipeline.repo_owner, pipeline.repo_name + ), + commit_sha: pipeline.commit_sha.clone(), + job_def: job.definition.clone(), + }; + + let runner = crate::runner::RunnerSupervisor::new( + ctx.self_addr(), + start_job, + ); + + match ctx.spawn(runner) { + Ok(runner_addr) => { + self.runner_addrs.insert(response.job_id, runner_addr); + } + Err(_) => { + // Failed to spawn runner — mark job as failed. + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + pipeline.set_job_status( + &job_name, + JobStatus::Failed { + reason: "failed to spawn runner".into(), + }, + ); + } + self.advance_pipeline(ctx, pipeline_id); + } + } + } + Err(err) => { + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + pipeline.set_job_status( + &job_name, + JobStatus::Failed { + reason: err.to_string(), + }, + ); + } + self.advance_pipeline(ctx, pipeline_id); + } + } + } + + fn handle_job_complete(&mut self, ctx: &Ctx, complete: JobComplete) { + let pipeline_id = complete.job_id.pipeline_id; + let job_name = complete.job_id.job_name.clone(); + + // Send terminate request for the instance. + if let Some(pipeline) = self.pipelines.get(&pipeline_id) { + if let Some(job) = pipeline.jobs.get(&job_name) { + if let Some(ref instance_id) = job.instance_id { + let terminate = TerminateRequest { + job_id: complete.job_id.clone(), + instance_id: instance_id.clone(), + }; + if let Some(prov_addr) = self.provisioner_addr { + let _ = ctx.send( + prov_addr, + crate::provisioner::ProvisionerMsg::Terminate(terminate), + ); + } + } + } + } + + // Update job status. + let status = match complete.result { + Ok(_) => JobStatus::Passed, + Err(ref failure) => JobStatus::Failed { + reason: failure.to_string(), + }, + }; + + let (repo_owner, repo_name, commit_sha) = { + let pipeline = match self.pipelines.get(&pipeline_id) { + Some(p) => p, + None => return, + }; + ( + pipeline.repo_owner.clone(), + pipeline.repo_name.clone(), + pipeline.commit_sha.clone(), + ) + }; + + // Emit per-job final status. + self.emit_status_update(StatusUpdate { + repo_owner, + repo_name, + commit_sha, + state: match &status { + JobStatus::Passed => "success".into(), + _ => "failure".into(), + }, + context: format!("ci/{job_name}"), + description: format!("Job '{job_name}' completed"), + }); + + if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) { + pipeline.set_job_status(&job_name, status); + } + + // Remove runner address. + self.runner_addrs.remove(&complete.job_id); + + // Advance pipeline to schedule downstream jobs. + self.advance_pipeline(ctx, pipeline_id); + } + + fn handle_provisioner_offline(&mut self) { + self.provisioner_online = false; + + // Mark all provisioning jobs as waiting. + for pipeline in self.pipelines.values_mut() { + for job in pipeline.jobs.values_mut() { + if job.status == JobStatus::Provisioning { + job.status = JobStatus::WaitingForProvisioner; + } + } + } + } + + fn handle_provisioner_online(&mut self, ctx: &Ctx) { + self.provisioner_online = true; + + // Flush queued provision requests. + let queued = std::mem::take(&mut self.queued_provisions); + for request in queued { + if let Some(prov_addr) = self.provisioner_addr { + let _ = ctx.send(prov_addr, crate::provisioner::ProvisionerMsg::Provision(request)); + } + } + + // Re-advance pipelines that have waiting jobs. + let pipeline_ids: Vec = self.pipelines.keys().copied().collect(); + for pid in pipeline_ids { + // Move waiting jobs back to provisioning. + if let Some(pipeline) = self.pipelines.get_mut(&pid) { + let waiting_jobs: Vec = pipeline + .jobs + .iter() + .filter(|(_, j)| j.status == JobStatus::WaitingForProvisioner) + .map(|(name, _)| name.clone()) + .collect(); + + for job_name in waiting_jobs { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.status = JobStatus::Pending; + } + } + } + self.advance_pipeline(ctx, pid); + } + } + + fn emit_status_update(&mut self, update: StatusUpdate) { + self.status_updates.push(update); + } +} + +impl ActorInterface for Coordinator { + type Incoming = CoordinatorMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: CoordinatorMsg) { + match msg { + CoordinatorMsg::Webhook(event) => self.handle_webhook(ctx, event), + CoordinatorMsg::SetCiYaml(yaml) => { + self.ci_yaml = Some(yaml); + } + CoordinatorMsg::ProvisionResponse(resp) => { + self.handle_provision_response(ctx, resp); + } + CoordinatorMsg::JobProgress(progress) => { + if let Some(pipeline) = self.pipelines.get_mut(&progress.job_id.pipeline_id) { + if let Some(job) = pipeline.jobs.get_mut(&progress.job_id.job_name) { + job.output_lines.push(progress.output_line); + } + } + } + CoordinatorMsg::JobComplete(complete) => { + self.handle_job_complete(ctx, complete); + } + CoordinatorMsg::ProvisionerOffline => self.handle_provisioner_offline(), + CoordinatorMsg::ProvisionerOnline => self.handle_provisioner_online(ctx), + } + } +} diff --git a/crates/ci/src/lib.rs b/crates/ci/src/lib.rs new file mode 100644 index 0000000..49925c9 --- /dev/null +++ b/crates/ci/src/lib.rs @@ -0,0 +1,288 @@ +pub mod coordinator; +pub mod pipeline; +pub mod provisioner; +pub mod runner; +pub mod yaml; + +use std::collections::HashMap; +use std::fmt; +use std::net::IpAddr; + +use serde::{Deserialize, Serialize}; + +// ─── Core Identifiers ─────────────────────────────────────────────────────── + +/// Unique identifier for a pipeline execution. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct PipelineId(pub u64); + +impl fmt::Display for PipelineId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "pipeline-{}", self.0) + } +} + +/// Unique identifier for a job within a pipeline. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct JobId { + pub pipeline_id: PipelineId, + pub job_name: String, +} + +impl fmt::Display for JobId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}/{}", self.pipeline_id, self.job_name) + } +} + +// ─── Job Status ───────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum JobStatus { + Pending, + WaitingForProvisioner, + Provisioning, + Running, + Passed, + Failed { reason: String }, + Skipped, + Interrupted, +} + +impl JobStatus { + pub fn is_terminal(&self) -> bool { + matches!( + self, + JobStatus::Passed | JobStatus::Failed { .. } | JobStatus::Skipped | JobStatus::Interrupted + ) + } +} + +// ─── Pipeline Status ──────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum PipelineStatus { + Pending, + Running, + Passed, + Failed, + Error { reason: String }, +} + +impl PipelineStatus { + pub fn is_terminal(&self) -> bool { + matches!( + self, + PipelineStatus::Passed | PipelineStatus::Failed | PipelineStatus::Error { .. } + ) + } + + /// Convert to Forgejo commit status string. + pub fn forgejo_state(&self) -> &'static str { + match self { + PipelineStatus::Pending => "pending", + PipelineStatus::Running => "pending", + PipelineStatus::Passed => "success", + PipelineStatus::Failed => "failure", + PipelineStatus::Error { .. } => "error", + } + } +} + +// ─── Instance Types ───────────────────────────────────────────────────────── + +/// Specification for a spot instance. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InstanceSpec { + pub min_cpus: u32, + pub min_ram_mb: u32, + pub min_disk_gb: u32, + pub docker_required: bool, + pub region_preferences: Vec, +} + +impl Default for InstanceSpec { + fn default() -> Self { + Self { + min_cpus: 2, + min_ram_mb: 2048, + min_disk_gb: 20, + docker_required: false, + region_preferences: Vec::new(), + } + } +} + +/// Connection details for a provisioned spot instance. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InstanceReady { + pub instance_id: String, + pub ip: IpAddr, + pub ssh_port: u16, + pub ssh_host_key: String, +} + +// ─── Provisioner ↔ Coordinator Messages ───────────────────────────────────── + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProvisionRequest { + pub job_id: JobId, + pub instance_spec: InstanceSpec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProvisionResponse { + pub job_id: JobId, + pub result: Result, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum ProvisionError { + NoCapacity, + ProviderError(String), + Timeout, + ProvisionerOffline, +} + +impl fmt::Display for ProvisionError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ProvisionError::NoCapacity => write!(f, "no capacity available"), + ProvisionError::ProviderError(msg) => write!(f, "provider error: {msg}"), + ProvisionError::Timeout => write!(f, "provisioning timed out"), + ProvisionError::ProvisionerOffline => write!(f, "provisioner is offline"), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TerminateRequest { + pub job_id: JobId, + pub instance_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TerminateAck { + pub job_id: JobId, +} + +// ─── Coordinator ↔ RunnerSupervisor Messages ──────────────────────────────── + +/// Sent from Coordinator to RunnerSupervisor to begin a job. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StartJob { + pub job_id: JobId, + pub instance: InstanceReady, + pub repo_url: String, + pub commit_sha: String, + pub job_def: JobDefinition, +} + +/// Streamed output from RunnerSupervisor back to Coordinator. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobProgress { + pub job_id: JobId, + pub output_line: String, +} + +/// Final result from RunnerSupervisor. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobComplete { + pub job_id: JobId, + pub result: Result, + pub artifacts: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobSuccess; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum JobFailure { + CommandFailed { exit_code: i32, last_lines: Vec }, + SshError(String), + Timeout, + Interrupted, +} + +impl fmt::Display for JobFailure { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + JobFailure::CommandFailed { exit_code, .. } => { + write!(f, "command exited with code {exit_code}") + } + JobFailure::SshError(msg) => write!(f, "SSH error: {msg}"), + JobFailure::Timeout => write!(f, "job timed out"), + JobFailure::Interrupted => write!(f, "spot instance interrupted"), + } + } +} + +// ─── Job Definition ───────────────────────────────────────────────────────── + +/// A parsed job from the .ci.yml file. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobDefinition { + pub name: String, + pub run: Vec, + pub needs: Vec, + pub timeout_secs: u64, + pub docker: bool, + pub artifacts: Vec, + pub env: HashMap, +} + +// ─── Webhook Types ────────────────────────────────────────────────────────── + +/// Parsed webhook event from Forgejo. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookEvent { + pub event_type: EventType, + pub repo_owner: String, + pub repo_name: String, + pub branch: String, + pub commit_sha: String, + pub tag: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum EventType { + Push, + Tag, + Merge, +} + +// ─── Forgejo Status Updates ───────────────────────────────────────────────── + +/// A commit status update to send to Forgejo's API. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StatusUpdate { + pub repo_owner: String, + pub repo_name: String, + pub commit_sha: String, + pub state: String, + pub context: String, + pub description: String, +} + +// ─── Coordinator Config ───────────────────────────────────────────────────── + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CiConfig { + pub webhook_port: u16, + pub webhook_secret: String, + pub forgejo_url: String, + pub forgejo_token: String, + pub data_dir: String, +} + +impl Default for CiConfig { + fn default() -> Self { + Self { + webhook_port: 8787, + webhook_secret: String::new(), + forgejo_url: String::new(), + forgejo_token: String::new(), + data_dir: "./ci-data".into(), + } + } +} diff --git a/crates/ci/src/pipeline.rs b/crates/ci/src/pipeline.rs new file mode 100644 index 0000000..20f241e --- /dev/null +++ b/crates/ci/src/pipeline.rs @@ -0,0 +1,391 @@ +//! Pipeline resolution, DAG execution logic, and job ordering. + +use std::collections::{HashMap, HashSet, VecDeque}; + +use crate::{JobDefinition, JobId, JobStatus, PipelineId, PipelineStatus}; + +/// A pipeline execution: tracks the DAG of jobs and their statuses. +#[derive(Debug, Clone)] +pub struct PipelineExecution { + pub pipeline_id: PipelineId, + pub pipeline_name: String, + pub repo_owner: String, + pub repo_name: String, + pub commit_sha: String, + pub branch: String, + pub status: PipelineStatus, + pub jobs: HashMap, +} + +/// State of a single job within a pipeline execution. +#[derive(Debug, Clone)] +pub struct JobExecution { + pub job_id: JobId, + pub definition: JobDefinition, + pub status: JobStatus, + pub output_lines: Vec, + pub instance_id: Option, +} + +impl PipelineExecution { + /// Create a new pipeline execution from a set of job definitions. + pub fn new( + pipeline_id: PipelineId, + pipeline_name: String, + repo_owner: String, + repo_name: String, + commit_sha: String, + branch: String, + job_defs: Vec, + ) -> Self { + let mut jobs = HashMap::new(); + for def in job_defs { + let job_id = JobId { + pipeline_id, + job_name: def.name.clone(), + }; + jobs.insert( + def.name.clone(), + JobExecution { + job_id, + definition: def, + status: JobStatus::Pending, + output_lines: Vec::new(), + instance_id: None, + }, + ); + } + Self { + pipeline_id, + pipeline_name, + repo_owner, + repo_name, + commit_sha, + branch, + status: PipelineStatus::Pending, + jobs, + } + } + + /// Return job names that are eligible for execution: Pending with all needs satisfied. + pub fn eligible_jobs(&self) -> Vec { + self.jobs + .values() + .filter(|job| { + job.status == JobStatus::Pending + && job.definition.needs.iter().all(|dep| { + self.jobs + .get(dep) + .map(|d| d.status == JobStatus::Passed) + .unwrap_or(false) + }) + }) + .map(|job| job.definition.name.clone()) + .collect() + } + + /// Mark a job as a given status. If a job fails, propagate skip to dependents. + pub fn set_job_status(&mut self, job_name: &str, status: JobStatus) { + if let Some(job) = self.jobs.get_mut(job_name) { + job.status = status.clone(); + } + + // If failure or interruption, skip all transitive dependents. + if matches!( + status, + JobStatus::Failed { .. } | JobStatus::Interrupted + ) { + let to_skip = self.transitive_dependents(job_name); + for dep_name in to_skip { + if let Some(dep_job) = self.jobs.get_mut(&dep_name) { + if dep_job.status == JobStatus::Pending { + dep_job.status = JobStatus::Skipped; + } + } + } + } + + // Update pipeline status. + self.update_pipeline_status(); + } + + /// Compute the overall pipeline status from individual job statuses. + fn update_pipeline_status(&mut self) { + let all_terminal = self.jobs.values().all(|j| j.status.is_terminal()); + let any_running = self.jobs.values().any(|j| { + matches!( + j.status, + JobStatus::Running | JobStatus::Provisioning | JobStatus::WaitingForProvisioner + ) + }); + let any_failed = self.jobs.values().any(|j| { + matches!( + j.status, + JobStatus::Failed { .. } | JobStatus::Interrupted + ) + }); + + if all_terminal { + self.status = if any_failed { + PipelineStatus::Failed + } else { + PipelineStatus::Passed + }; + } else if any_running + || self + .jobs + .values() + .any(|j| j.status == JobStatus::Passed) + { + self.status = PipelineStatus::Running; + } + } + + /// Find all transitive dependents of a job (jobs that directly or indirectly need it). + fn transitive_dependents(&self, job_name: &str) -> Vec { + let mut result = Vec::new(); + let mut queue: VecDeque<&str> = VecDeque::new(); + queue.push_back(job_name); + let mut visited = HashSet::new(); + + while let Some(current) = queue.pop_front() { + for (name, job) in &self.jobs { + if job.definition.needs.iter().any(|n| n == current) && visited.insert(name.clone()) + { + result.push(name.clone()); + queue.push_back(name.as_str()); + } + } + } + result + } +} + +// ─── DAG Validation ───────────────────────────────────────────────────────── + +/// Topological sort of job definitions. Returns ordered job names or an error if cyclic. +pub fn topological_sort(jobs: &HashMap) -> Result, DagError> { + let mut in_degree: HashMap<&str, usize> = HashMap::new(); + let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new(); + + for (name, def) in jobs { + in_degree.entry(name.as_str()).or_insert(0); + for dep in &def.needs { + if !jobs.contains_key(dep) { + return Err(DagError::MissingDependency { + job: name.clone(), + dependency: dep.clone(), + }); + } + dependents.entry(dep.as_str()).or_default().push(name.as_str()); + *in_degree.entry(name.as_str()).or_insert(0) += 1; + } + } + + let mut queue: VecDeque<&str> = in_degree + .iter() + .filter(|(_, deg)| **deg == 0) + .map(|(&name, _)| name) + .collect(); + + // Sort the initial queue for deterministic ordering. + let mut sorted_queue: Vec<&str> = queue.drain(..).collect(); + sorted_queue.sort(); + queue.extend(sorted_queue); + + let mut result = Vec::new(); + while let Some(node) = queue.pop_front() { + result.push(node.to_string()); + if let Some(deps) = dependents.get(node) { + let mut next_nodes = Vec::new(); + for &dep in deps { + if let Some(deg) = in_degree.get_mut(dep) { + *deg -= 1; + if *deg == 0 { + next_nodes.push(dep); + } + } + } + next_nodes.sort(); + queue.extend(next_nodes); + } + } + + if result.len() != jobs.len() { + return Err(DagError::Cycle); + } + Ok(result) +} + +#[derive(Debug, Clone)] +pub enum DagError { + Cycle, + MissingDependency { job: String, dependency: String }, +} + +impl std::fmt::Display for DagError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + DagError::Cycle => write!(f, "job dependency cycle detected"), + DagError::MissingDependency { job, dependency } => { + write!(f, "job '{job}' depends on unknown job '{dependency}'") + } + } + } +} + +impl std::error::Error for DagError {} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + use crate::JobDefinition; + + fn make_job(name: &str, needs: &[&str]) -> JobDefinition { + JobDefinition { + name: name.into(), + run: vec!["echo test".into()], + needs: needs.iter().map(|s| s.to_string()).collect(), + timeout_secs: 300, + docker: false, + artifacts: Vec::new(), + env: HashMap::new(), + } + } + + #[test] + fn topological_sort_linear_chain() { + let mut jobs = HashMap::new(); + jobs.insert("a".into(), make_job("a", &[])); + jobs.insert("b".into(), make_job("b", &["a"])); + jobs.insert("c".into(), make_job("c", &["b"])); + + let order = topological_sort(&jobs).unwrap(); + assert_eq!(order, vec!["a", "b", "c"]); + } + + #[test] + fn topological_sort_diamond() { + let mut jobs = HashMap::new(); + jobs.insert("a".into(), make_job("a", &[])); + jobs.insert("b".into(), make_job("b", &["a"])); + jobs.insert("c".into(), make_job("c", &["a"])); + jobs.insert("d".into(), make_job("d", &["b", "c"])); + + let order = topological_sort(&jobs).unwrap(); + let pos = |name: &str| order.iter().position(|n| n == name).unwrap(); + assert!(pos("a") < pos("b")); + assert!(pos("a") < pos("c")); + assert!(pos("b") < pos("d")); + assert!(pos("c") < pos("d")); + } + + #[test] + fn topological_sort_detects_cycle() { + let mut jobs = HashMap::new(); + jobs.insert("a".into(), make_job("a", &["b"])); + jobs.insert("b".into(), make_job("b", &["a"])); + + let result = topological_sort(&jobs); + assert!(matches!(result, Err(DagError::Cycle))); + } + + #[test] + fn topological_sort_independent_jobs() { + let mut jobs = HashMap::new(); + jobs.insert("a".into(), make_job("a", &[])); + jobs.insert("b".into(), make_job("b", &[])); + jobs.insert("c".into(), make_job("c", &[])); + + let order = topological_sort(&jobs).unwrap(); + // All jobs present, order is alphabetical for independent nodes + assert_eq!(order.len(), 3); + assert_eq!(order, vec!["a", "b", "c"]); + } + + #[test] + fn pipeline_eligible_jobs_respects_dag() { + let jobs = vec![ + make_job("fmt", &[]), + make_job("clippy", &[]), + make_job("test", &["fmt", "clippy"]), + ]; + + let mut pipeline = PipelineExecution::new( + PipelineId(1), + "check".into(), + "user".into(), + "repo".into(), + "abc123".into(), + "main".into(), + jobs, + ); + + // Initially, fmt and clippy are eligible + let mut eligible = pipeline.eligible_jobs(); + eligible.sort(); + assert_eq!(eligible, vec!["clippy", "fmt"]); + + // After fmt passes, test is still not eligible (clippy pending) + pipeline.set_job_status("fmt", JobStatus::Passed); + let eligible = pipeline.eligible_jobs(); + assert_eq!(eligible, vec!["clippy"]); + + // After clippy passes, test becomes eligible + pipeline.set_job_status("clippy", JobStatus::Passed); + let eligible = pipeline.eligible_jobs(); + assert_eq!(eligible, vec!["test"]); + } + + #[test] + fn pipeline_failure_skips_dependents() { + let jobs = vec![ + make_job("fmt", &[]), + make_job("test", &["fmt"]), + make_job("bench", &["test"]), + ]; + + let mut pipeline = PipelineExecution::new( + PipelineId(1), + "check".into(), + "user".into(), + "repo".into(), + "abc123".into(), + "main".into(), + jobs, + ); + + // fmt fails → test and bench should be skipped + pipeline.set_job_status( + "fmt", + JobStatus::Failed { + reason: "formatting error".into(), + }, + ); + + assert_eq!(pipeline.jobs["test"].status, JobStatus::Skipped); + assert_eq!(pipeline.jobs["bench"].status, JobStatus::Skipped); + assert_eq!(pipeline.status, PipelineStatus::Failed); + } + + #[test] + fn pipeline_all_pass_yields_passed() { + let jobs = vec![make_job("a", &[]), make_job("b", &["a"])]; + + let mut pipeline = PipelineExecution::new( + PipelineId(1), + "check".into(), + "user".into(), + "repo".into(), + "abc123".into(), + "main".into(), + jobs, + ); + + pipeline.set_job_status("a", JobStatus::Passed); + pipeline.set_job_status("b", JobStatus::Passed); + assert_eq!(pipeline.status, PipelineStatus::Passed); + } +} diff --git a/crates/ci/src/provisioner.rs b/crates/ci/src/provisioner.rs new file mode 100644 index 0000000..26b007c --- /dev/null +++ b/crates/ci/src/provisioner.rs @@ -0,0 +1,86 @@ +//! Provisioner actor: manages spot instance lifecycle via pluggable provider scripts. +//! +//! In real deployment, runs on the developer's laptop and calls cloud provider APIs. +//! In simulation, provisions are driven by the simulation harness. + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use crate::coordinator::CoordinatorMsg; +use crate::{ + InstanceReady, ProvisionError, ProvisionRequest, ProvisionResponse, TerminateRequest, +}; + +/// Messages the Provisioner can receive. +#[derive(Debug, Clone)] +pub enum ProvisionerMsg { + /// Request to provision a new spot instance. + Provision(ProvisionRequest), + /// Request to terminate a spot instance. + Terminate(TerminateRequest), + /// Simulated: provisioning result delivered asynchronously. + SimProvisionResult { + request: ProvisionRequest, + result: Result, + }, +} + +/// Provisioner actor state. +/// +/// In real deployment, this would invoke provider scripts. +/// In simulation, the sim harness controls provision outcomes. +pub struct Provisioner { + coordinator_addr: ActorAddress, + /// Active instances tracked for cleanup. + active_instances: Vec, +} + +impl Provisioner { + pub fn new(coordinator_addr: ActorAddress) -> Self { + Self { + coordinator_addr, + active_instances: Vec::new(), + } + } + + pub fn active_instances(&self) -> &[String] { + &self.active_instances + } +} + +impl ActorInterface for Provisioner { + type Incoming = ProvisionerMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: ProvisionerMsg) { + match msg { + ProvisionerMsg::Provision(_request) => { + // In real deployment: invoke provider script, await result. + // In simulation: the sim harness sends SimProvisionResult. + } + ProvisionerMsg::Terminate(request) => { + self.active_instances.retain(|id| id != &request.instance_id); + // In real deployment: invoke provider destroy script. + // In simulation: just track the termination. + let _ = ctx.send( + self.coordinator_addr, + CoordinatorMsg::ProvisionResponse(ProvisionResponse { + job_id: request.job_id.clone(), + result: Err(ProvisionError::NoCapacity), // placeholder, terminate doesn't need response + }), + ); + } + ProvisionerMsg::SimProvisionResult { request, result } => { + if let Ok(ref instance) = result { + self.active_instances.push(instance.instance_id.clone()); + } + let _ = ctx.send( + self.coordinator_addr, + CoordinatorMsg::ProvisionResponse(ProvisionResponse { + job_id: request.job_id, + result, + }), + ); + } + } + } +} diff --git a/crates/ci/src/runner.rs b/crates/ci/src/runner.rs new file mode 100644 index 0000000..99c0357 --- /dev/null +++ b/crates/ci/src/runner.rs @@ -0,0 +1,84 @@ +//! RunnerSupervisor actor: manages SSH session and job execution on a spot instance. +//! +//! Spawned per-job by the Coordinator. Owns the connection to the spot instance. + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use crate::coordinator::CoordinatorMsg; +use crate::{JobComplete, JobFailure, JobProgress, JobSuccess, StartJob}; + +/// Messages the RunnerSupervisor can receive. +#[derive(Debug, Clone)] +pub enum RunnerMsg { + /// Begin executing the job (sent immediately after spawn via on_start). + Execute, + /// Simulated: job command output line. + OutputLine(String), + /// Simulated: job completed successfully. + SimComplete(Result<(), String>), +} + +/// RunnerSupervisor actor state. +/// +/// In real deployment, this would manage an SSH connection. +/// In simulation, job execution is driven by external messages. +pub struct RunnerSupervisor { + coordinator_addr: ActorAddress, + start_job: StartJob, +} + +impl RunnerSupervisor { + pub fn new(coordinator_addr: ActorAddress, start_job: StartJob) -> Self { + Self { + coordinator_addr, + start_job, + } + } +} + +impl ActorInterface for RunnerSupervisor { + type Incoming = RunnerMsg; + type Response = (); + + fn on_start(&mut self, ctx: &Ctx) { + // In simulation, the sim harness will send SimComplete messages. + // In real deployment, this would initiate SSH connection + command execution. + let _ = ctx.send(ctx.self_addr(), RunnerMsg::Execute); + } + + fn handle(&mut self, ctx: &Ctx, msg: RunnerMsg) { + match msg { + RunnerMsg::Execute => { + // In real mode, we'd SSH into the instance and run commands. + // In simulation, this is a no-op; SimComplete drives completion. + } + RunnerMsg::OutputLine(line) => { + let _ = ctx.send( + self.coordinator_addr, + CoordinatorMsg::JobProgress(JobProgress { + job_id: self.start_job.job_id.clone(), + output_line: line, + }), + ); + } + RunnerMsg::SimComplete(result) => { + let complete = JobComplete { + job_id: self.start_job.job_id.clone(), + result: match result { + Ok(()) => Ok(JobSuccess), + Err(msg) => Err(JobFailure::CommandFailed { + exit_code: 1, + last_lines: vec![msg], + }), + }, + artifacts: Vec::new(), + }; + let _ = ctx.send( + self.coordinator_addr, + CoordinatorMsg::JobComplete(complete), + ); + ctx.stop_self(); + } + } + } +} diff --git a/crates/ci/src/yaml.rs b/crates/ci/src/yaml.rs new file mode 100644 index 0000000..de50472 --- /dev/null +++ b/crates/ci/src/yaml.rs @@ -0,0 +1,432 @@ +//! Parser for `.ci.yml` pipeline configuration files. + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +use crate::{EventType, JobDefinition, WebhookEvent}; + +/// Root of a `.ci.yml` file. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CiYaml { + pub pipelines: HashMap, +} + +/// A single pipeline definition. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineDef { + pub triggers: Vec, + pub jobs: HashMap, +} + +/// A trigger condition. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TriggerDef { + pub event: TriggerEvent, + #[serde(default)] + pub branches: Vec, + #[serde(default)] + pub exclude: Vec, + #[serde(default)] + pub pattern: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum TriggerEvent { + Push, + Tag, + Merge, +} + +/// A job definition in YAML form. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobDef { + pub run: RunCommand, + #[serde(default)] + pub needs: Vec, + #[serde(default)] + pub timeout: Option, + #[serde(default)] + pub docker: bool, + #[serde(default)] + pub artifacts: Vec, + #[serde(default)] + pub env: HashMap, +} + +/// `run` can be a single string or an array of strings. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum RunCommand { + Single(String), + Multiple(Vec), +} + +impl RunCommand { + pub fn into_vec(self) -> Vec { + match self { + RunCommand::Single(s) => vec![s], + RunCommand::Multiple(v) => v, + } + } +} + +// ─── Parsing ──────────────────────────────────────────────────────────────── + +/// Parse a `.ci.yml` string into a `CiYaml`. +pub fn parse_ci_yaml(input: &str) -> Result { + let ci: CiYaml = serde_yaml::from_str(input).map_err(ParseError::Yaml)?; + + // Validate: check for unknown job references in `needs` + for (pipeline_name, pipeline) in &ci.pipelines { + let job_names: Vec<&str> = pipeline.jobs.keys().map(|s| s.as_str()).collect(); + for (job_name, job) in &pipeline.jobs { + for dep in &job.needs { + if !job_names.contains(&dep.as_str()) { + return Err(ParseError::UnknownDependency { + pipeline: pipeline_name.clone(), + job: job_name.clone(), + dependency: dep.clone(), + }); + } + } + } + } + + Ok(ci) +} + +/// Convert a YAML `JobDef` to the runtime `JobDefinition`. +pub fn to_job_definition(name: &str, def: &JobDef) -> JobDefinition { + JobDefinition { + name: name.to_string(), + run: def.run.clone().into_vec(), + needs: def.needs.clone(), + timeout_secs: def.timeout.unwrap_or(300), + docker: def.docker, + artifacts: def.artifacts.clone(), + env: def.env.clone(), + } +} + +// ─── Trigger Matching ─────────────────────────────────────────────────────── + +/// Returns the names of pipelines whose triggers match the given webhook event. +pub fn matching_pipelines(ci: &CiYaml, event: &WebhookEvent) -> Vec { + ci.pipelines + .iter() + .filter(|(_, pipeline)| pipeline.triggers.iter().any(|t| trigger_matches(t, event))) + .map(|(name, _)| name.clone()) + .collect() +} + +/// Check whether a single trigger matches a webhook event. +fn trigger_matches(trigger: &TriggerDef, event: &WebhookEvent) -> bool { + // Event type must match. + let event_matches = match (&trigger.event, &event.event_type) { + (TriggerEvent::Push, EventType::Push) => true, + (TriggerEvent::Tag, EventType::Tag) => true, + (TriggerEvent::Merge, EventType::Merge) => true, + _ => false, + }; + if !event_matches { + return false; + } + + // For tag events, check pattern. + if trigger.event == TriggerEvent::Tag { + if let Some(ref pattern) = trigger.pattern { + return glob_matches(pattern, event.tag.as_deref().unwrap_or("")); + } + return true; + } + + // For push/merge, check branch filters. + let branch = &event.branch; + + // If exclude patterns are specified and branch matches any, reject. + if trigger.exclude.iter().any(|pat| glob_matches(pat, branch)) { + return false; + } + + // If branch patterns are specified, at least one must match. + if trigger.branches.is_empty() { + return true; + } + trigger.branches.iter().any(|pat| glob_matches(pat, branch)) +} + +/// Simple glob matching supporting `*` (any chars) and `?` (one char). +pub fn glob_matches(pattern: &str, text: &str) -> bool { + glob_matches_inner(pattern.as_bytes(), text.as_bytes()) +} + +fn glob_matches_inner(pat: &[u8], text: &[u8]) -> bool { + match (pat.first(), text.first()) { + (None, None) => true, + (Some(b'*'), _) => { + // '*' matches zero or more characters + glob_matches_inner(&pat[1..], text) + || (!text.is_empty() && glob_matches_inner(pat, &text[1..])) + } + (Some(b'?'), Some(_)) => glob_matches_inner(&pat[1..], &text[1..]), + (Some(a), Some(b)) if a == b => glob_matches_inner(&pat[1..], &text[1..]), + _ => false, + } +} + +// ─── Errors ───────────────────────────────────────────────────────────────── + +#[derive(Debug)] +pub enum ParseError { + Yaml(serde_yaml::Error), + UnknownDependency { + pipeline: String, + job: String, + dependency: String, + }, +} + +impl std::fmt::Display for ParseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ParseError::Yaml(e) => write!(f, "YAML parse error: {e}"), + ParseError::UnknownDependency { + pipeline, + job, + dependency, + } => write!( + f, + "pipeline '{pipeline}', job '{job}': unknown dependency '{dependency}'" + ), + } + } +} + +impl std::error::Error for ParseError {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_minimal_ci_yaml() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + test: + run: cargo test +"#; + let ci = parse_ci_yaml(yaml).unwrap(); + assert_eq!(ci.pipelines.len(), 1); + assert!(ci.pipelines.contains_key("check")); + let check = &ci.pipelines["check"]; + assert_eq!(check.jobs.len(), 1); + assert!(check.jobs.contains_key("test")); + } + + #[test] + fn parse_full_ci_yaml() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + exclude: ["master"] + jobs: + fmt: + run: cargo fmt -- --check + clippy: + run: cargo clippy --all-features -- -D warnings + test: + needs: [fmt, clippy] + run: cargo test + full: + triggers: + - event: push + branches: ["master"] + jobs: + test: + run: cargo test --all-features + timeout: 600 + bench: + needs: [test] + run: cargo bench -- --output-format json + artifacts: ["target/criterion/**"] + docker: + needs: [test] + run: cargo test -p docker-tests -- --ignored + docker: true + release: + triggers: + - event: tag + pattern: "v*" + jobs: + build: + run: cargo build --release + artifacts: ["target/release/swactor-node"] +"#; + let ci = parse_ci_yaml(yaml).unwrap(); + assert_eq!(ci.pipelines.len(), 3); + assert!(ci.pipelines.contains_key("check")); + assert!(ci.pipelines.contains_key("full")); + assert!(ci.pipelines.contains_key("release")); + + let full = &ci.pipelines["full"]; + assert_eq!(full.jobs.len(), 3); + assert_eq!(full.jobs["bench"].needs, vec!["test"]); + assert!(full.jobs["docker"].docker); + } + + #[test] + fn unknown_dependency_rejected() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + test: + needs: [nonexistent] + run: cargo test +"#; + let result = parse_ci_yaml(yaml); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!( + matches!(err, ParseError::UnknownDependency { .. }), + "expected UnknownDependency, got: {err}" + ); + } + + #[test] + fn trigger_matching_push_branch() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + exclude: ["master"] + jobs: + test: + run: cargo test + full: + triggers: + - event: push + branches: ["master"] + jobs: + test: + run: cargo test +"#; + let ci = parse_ci_yaml(yaml).unwrap(); + + // Push to feature branch → matches "check" only + let event = WebhookEvent { + event_type: EventType::Push, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: "feature-x".into(), + commit_sha: "abc123".into(), + tag: None, + }; + let mut matched = matching_pipelines(&ci, &event); + matched.sort(); + assert_eq!(matched, vec!["check"]); + + // Push to master → matches "full" only (check excludes master) + let event = WebhookEvent { + event_type: EventType::Push, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: "master".into(), + commit_sha: "abc123".into(), + tag: None, + }; + let mut matched = matching_pipelines(&ci, &event); + matched.sort(); + assert_eq!(matched, vec!["full"]); + } + + #[test] + fn trigger_matching_tag() { + let yaml = r#" +pipelines: + release: + triggers: + - event: tag + pattern: "v*" + jobs: + build: + run: cargo build --release +"#; + let ci = parse_ci_yaml(yaml).unwrap(); + + let event = WebhookEvent { + event_type: EventType::Tag, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: "master".into(), + commit_sha: "abc123".into(), + tag: Some("v1.0.0".into()), + }; + let matched = matching_pipelines(&ci, &event); + assert_eq!(matched, vec!["release"]); + + // Non-matching tag + let event = WebhookEvent { + event_type: EventType::Tag, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: "master".into(), + commit_sha: "abc123".into(), + tag: Some("nightly-1".into()), + }; + let matched = matching_pipelines(&ci, &event); + assert!(matched.is_empty()); + } + + #[test] + fn glob_matching() { + assert!(glob_matches("*", "anything")); + assert!(glob_matches("v*", "v1.0.0")); + assert!(!glob_matches("v*", "nightly")); + assert!(glob_matches("feature-?", "feature-x")); + assert!(!glob_matches("feature-?", "feature-xy")); + assert!(glob_matches("master", "master")); + assert!(!glob_matches("master", "main")); + } + + #[test] + fn run_command_single_and_multiple() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + single: + run: cargo test + multi: + run: + - cargo fmt -- --check + - cargo test +"#; + let ci = parse_ci_yaml(yaml).unwrap(); + let check = &ci.pipelines["check"]; + + let single = to_job_definition("single", &check.jobs["single"]); + assert_eq!(single.run, vec!["cargo test"]); + + let multi = to_job_definition("multi", &check.jobs["multi"]); + assert_eq!(multi.run, vec!["cargo fmt -- --check", "cargo test"]); + } +} diff --git a/crates/runtime-dashboard/Cargo.toml b/crates/runtime-dashboard/Cargo.toml index 76fde7d..4556bd8 100644 --- a/crates/runtime-dashboard/Cargo.toml +++ b/crates/runtime-dashboard/Cargo.toml @@ -17,6 +17,7 @@ distribution = { path = "../distribution", optional = true } clap = { version = "4", features = ["derive"], optional = true } ctrlc = "3" iroh = { version = "0.96", optional = true } +swactor-ci = { path = "../ci", optional = true } [features] default = ["distribution"] @@ -25,6 +26,7 @@ distribution = ["dep:distribution"] node = ["distribution", "dep:clap", "swactor/transport", "tcp"] tcp = ["distribution/tcp"] iroh = ["distribution/iroh", "dep:iroh"] +ci = ["dep:swactor-ci"] [[bin]] name = "swactor-tui" diff --git a/crates/runtime-dashboard/src/ci_collector.rs b/crates/runtime-dashboard/src/ci_collector.rs new file mode 100644 index 0000000..85b321e --- /dev/null +++ b/crates/runtime-dashboard/src/ci_collector.rs @@ -0,0 +1,210 @@ +//! CI Dashboard extension: provides HTTP API endpoints and stats for CI pipelines. + +use serde::{Deserialize, Serialize}; + +use swactor_ci::{JobId, JobStatus, PipelineId, PipelineStatus}; + +// ─── Stats Provider ───────────────────────────────────────────────────────── + +/// Trait for providing CI snapshot data to the dashboard. +pub trait CiStatsProvider: Send + Sync { + fn snapshot(&self) -> CiSnapshot; +} + +/// Point-in-time snapshot of CI system state. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CiSnapshot { + pub active_pipelines: Vec, + pub recent_pipelines: Vec, + pub provisioner_status: ProvisionerStatus, + pub active_instances: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum ProvisionerStatus { + Online, + Offline, + Unknown, +} + +/// Snapshot of a single pipeline execution. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineSnapshot { + pub pipeline_id: PipelineId, + pub pipeline_name: String, + pub repo_owner: String, + pub repo_name: String, + pub commit_sha: String, + pub branch: String, + pub status: PipelineStatus, + pub jobs: Vec, +} + +/// Snapshot of a single job within a pipeline. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobSnapshot { + pub job_id: JobId, + pub job_name: String, + pub status: JobStatus, + pub output_line_count: usize, +} + +// ─── HTTP API Responses ───────────────────────────────────────────────────── + +/// Response for GET /api/ci/pipelines +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineListResponse { + pub pipelines: Vec, +} + +/// Response for GET /api/ci/pipelines/{id} +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineDetailResponse { + pub pipeline: PipelineSnapshot, +} + +/// Response for GET /api/ci/status +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SystemStatusResponse { + pub provisioner_status: ProvisionerStatus, + pub active_pipelines: usize, + pub active_instances: usize, +} + +/// Response for GET /api/ci/pipelines/{id}/jobs/{job_id}/log +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobLogResponse { + pub job_id: JobId, + pub lines: Vec, +} + +// ─── Route Matching ───────────────────────────────────────────────────────── + +/// Parsed API route for the CI dashboard. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CiRoute { + ListPipelines, + PipelineDetail { id: u64 }, + JobLog { pipeline_id: u64, job_name: String }, + Artifact { job_id: String, path: String }, + SystemStatus, + NotFound, +} + +/// Parse a request path into a CiRoute. +pub fn parse_route(path: &str) -> CiRoute { + let parts: Vec<&str> = path.trim_start_matches('/').split('/').collect(); + match parts.as_slice() { + ["api", "ci", "pipelines"] => CiRoute::ListPipelines, + ["api", "ci", "pipelines", id] => { + if let Ok(id) = id.parse() { + CiRoute::PipelineDetail { id } + } else { + CiRoute::NotFound + } + } + ["api", "ci", "pipelines", id, "jobs", job_name, "log"] => { + if let Ok(pipeline_id) = id.parse() { + CiRoute::JobLog { + pipeline_id, + job_name: job_name.to_string(), + } + } else { + CiRoute::NotFound + } + } + ["api", "ci", "artifacts", job_id, rest @ ..] if !rest.is_empty() => CiRoute::Artifact { + job_id: job_id.to_string(), + path: rest.join("/"), + }, + ["api", "ci", "status"] => CiRoute::SystemStatus, + _ => CiRoute::NotFound, + } +} + +/// Render a CiSnapshot into a JSON response for the given route. +pub fn handle_route(route: &CiRoute, snapshot: &CiSnapshot) -> Option { + match route { + CiRoute::ListPipelines => { + let resp = PipelineListResponse { + pipelines: snapshot + .active_pipelines + .iter() + .chain(snapshot.recent_pipelines.iter()) + .cloned() + .collect(), + }; + serde_json::to_string(&resp).ok() + } + CiRoute::PipelineDetail { id } => { + let pipeline = snapshot + .active_pipelines + .iter() + .chain(snapshot.recent_pipelines.iter()) + .find(|p| p.pipeline_id.0 == *id)?; + let resp = PipelineDetailResponse { + pipeline: pipeline.clone(), + }; + serde_json::to_string(&resp).ok() + } + CiRoute::SystemStatus => { + let resp = SystemStatusResponse { + provisioner_status: snapshot.provisioner_status.clone(), + active_pipelines: snapshot.active_pipelines.len(), + active_instances: snapshot.active_instances, + }; + serde_json::to_string(&resp).ok() + } + CiRoute::JobLog { .. } | CiRoute::Artifact { .. } => { + // These require access to stored data beyond the snapshot. + None + } + CiRoute::NotFound => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn route_parsing() { + assert_eq!(parse_route("/api/ci/pipelines"), CiRoute::ListPipelines); + assert_eq!( + parse_route("/api/ci/pipelines/42"), + CiRoute::PipelineDetail { id: 42 } + ); + assert_eq!( + parse_route("/api/ci/pipelines/1/jobs/test/log"), + CiRoute::JobLog { + pipeline_id: 1, + job_name: "test".into() + } + ); + assert_eq!( + parse_route("/api/ci/artifacts/job-1/target/release/bin"), + CiRoute::Artifact { + job_id: "job-1".into(), + path: "target/release/bin".into() + } + ); + assert_eq!(parse_route("/api/ci/status"), CiRoute::SystemStatus); + assert_eq!(parse_route("/api/ci/unknown"), CiRoute::NotFound); + } + + #[test] + fn handle_system_status() { + let snapshot = CiSnapshot { + active_pipelines: vec![], + recent_pipelines: vec![], + provisioner_status: ProvisionerStatus::Online, + active_instances: 2, + }; + + let route = CiRoute::SystemStatus; + let json = handle_route(&route, &snapshot).unwrap(); + let resp: SystemStatusResponse = serde_json::from_str(&json).unwrap(); + assert_eq!(resp.provisioner_status, ProvisionerStatus::Online); + assert_eq!(resp.active_instances, 2); + } +} diff --git a/crates/runtime-dashboard/src/lib.rs b/crates/runtime-dashboard/src/lib.rs index 1db0871..bc0b5a2 100644 --- a/crates/runtime-dashboard/src/lib.rs +++ b/crates/runtime-dashboard/src/lib.rs @@ -23,6 +23,9 @@ pub mod distribution_collector; mod datastore_html; pub mod datastore_collector; +#[cfg(feature = "ci")] +pub mod ci_collector; + use std::io; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -95,6 +98,8 @@ pub struct DashboardHandle { #[cfg(feature = "distribution")] distribution: Arc>>>, datastore: Arc>>>, + #[cfg(feature = "ci")] + ci: Arc>>>, } impl DashboardHandle { @@ -132,6 +137,12 @@ impl DashboardHandle { *self.datastore.lock().unwrap() = Some(provider); } + /// Attach a CI stats provider, enabling the `/api/ci/*` endpoints. + #[cfg(feature = "ci")] + pub fn set_ci(&self, provider: Arc) { + *self.ci.lock().unwrap() = Some(provider); + } + /// Access the time-series history store (for TUI sparklines, etc.). pub fn history(&self) -> &Arc { &self.history @@ -190,6 +201,10 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { let datastore: Arc>>> = Arc::new(Mutex::new(None)); + #[cfg(feature = "ci")] + let ci: Arc>>> = + Arc::new(Mutex::new(None)); + server::spawn_http_server( Arc::clone(&store), Arc::clone(&runtime), @@ -200,6 +215,8 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { #[cfg(feature = "distribution")] Arc::clone(&distribution), Arc::clone(&datastore), + #[cfg(feature = "ci")] + Arc::clone(&ci), ); // Start stats recorder thread when recording is enabled @@ -243,6 +260,8 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle { #[cfg(feature = "distribution")] distribution, datastore, + #[cfg(feature = "ci")] + ci, } } diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs index ee95b19..d7d369d 100644 --- a/crates/runtime-dashboard/src/server.rs +++ b/crates/runtime-dashboard/src/server.rs @@ -27,6 +27,9 @@ use crate::distribution_html::DISTRIBUTION_HTML; use crate::datastore_collector::DatastoreStatsProvider; use crate::datastore_html::DATASTORE_HTML; +#[cfg(feature = "ci")] +use crate::ci_collector::CiStatsProvider; + /// Format a server-sent event. fn format_sse(event: &str, data: &str) -> Vec { format!("event: {event}\ndata: {data}\n\n").into_bytes() @@ -132,6 +135,8 @@ pub(crate) fn spawn_http_server( #[cfg(feature = "distribution")] distribution: Arc>>>, datastore: Arc>>>, + #[cfg(feature = "ci")] + ci: Arc>>>, ) { let addr = format!("0.0.0.0:{port}"); let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); @@ -149,6 +154,8 @@ pub(crate) fn spawn_http_server( #[cfg(feature = "distribution")] let distribution = Arc::clone(&distribution); let datastore = Arc::clone(&datastore); + #[cfg(feature = "ci")] + let ci = Arc::clone(&ci); thread::spawn(move || { loop { let request = match server.recv() { @@ -176,6 +183,8 @@ pub(crate) fn spawn_http_server( #[cfg(feature = "distribution")] Arc::clone(&distribution), Arc::clone(&datastore), + #[cfg(feature = "ci")] + Arc::clone(&ci), ); } "/api/stats" => { @@ -220,6 +229,10 @@ pub(crate) fn spawn_http_server( "/api/logs" => { handle_logs_api(request, &url, Arc::clone(&store)); } + #[cfg(feature = "ci")] + _ if path.starts_with("/api/ci/") => { + handle_ci_api(request, path, Arc::clone(&ci)); + } _ if path.starts_with("/actor/") => { let hex = &path[7..]; // strip "/actor/" respond_actor_detail(request, hex); @@ -253,6 +266,8 @@ fn handle_live_sse( #[cfg(feature = "distribution")] distribution: Arc>>>, datastore: Arc>>>, + #[cfg(feature = "ci")] + ci: Arc>>>, ) { let (tx, rx) = mpsc::channel::>(); let response = make_sse_response(rx); @@ -335,6 +350,20 @@ fn handle_live_sse( } } + // Send CI snapshot if provider is attached + #[cfg(feature = "ci")] + { + let maybe_ci = ci.lock().unwrap().clone(); + if let Some(provider) = maybe_ci { + let snapshot = provider.snapshot(); + if let Ok(json) = serde_json::to_string(&snapshot) { + if tx.send(format_sse("ci", &json)).is_err() { + return; + } + } + } + } + // Send new activity events let (batch, new_cursor) = store.read_from(cursor); if !batch.is_empty() { @@ -469,6 +498,35 @@ fn handle_datastore_api( let _ = request.respond(response); } +#[cfg(feature = "ci")] +fn handle_ci_api( + request: tiny_http::Request, + path: &str, + ci: Arc>>>, +) { + use crate::ci_collector; + + let route = ci_collector::parse_route(path); + let json = match ci.lock().unwrap().as_ref() { + Some(provider) => { + let snapshot = provider.snapshot(); + ci_collector::handle_route(&route, &snapshot) + .unwrap_or_else(|| r#"{"error":"not found"}"#.to_string()) + } + None => serde_json::json!({ + "error": "CI provider not attached" + }) + .to_string(), + }; + + let response = tiny_http::Response::from_string(json).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + fn handle_topology_api( request: tiny_http::Request, runtime: Arc>>>, diff --git a/crates/simulation/Cargo.toml b/crates/simulation/Cargo.toml index 93c8b35..6f77f91 100644 --- a/crates/simulation/Cargo.toml +++ b/crates/simulation/Cargo.toml @@ -7,6 +7,7 @@ edition = "2024" default = [] gossip = ["dep:log"] dashboard = ["gossip", "dep:tiny_http", "dep:toml"] +ci = ["dep:swactor-ci"] [dependencies] distribution = { path = "../distribution" } @@ -17,9 +18,11 @@ getrandom = "0.2" log = { version = "0.4", optional = true } tiny_http = { version = "0.12", optional = true } toml = { version = "0.8", optional = true } +swactor-ci = { path = "../ci", optional = true } [dev-dependencies] -simulation = { path = ".", features = ["gossip"] } +simulation = { path = ".", features = ["gossip", "ci"] } +swactor-ci = { path = "../ci" } [[example]] name = "gossip_sim" diff --git a/crates/simulation/src/ci/mod.rs b/crates/simulation/src/ci/mod.rs new file mode 100644 index 0000000..e722c10 --- /dev/null +++ b/crates/simulation/src/ci/mod.rs @@ -0,0 +1 @@ +pub mod sim; diff --git a/crates/simulation/src/ci/sim.rs b/crates/simulation/src/ci/sim.rs new file mode 100644 index 0000000..0158018 --- /dev/null +++ b/crates/simulation/src/ci/sim.rs @@ -0,0 +1,610 @@ +//! CI protocol simulation: deterministic, round-based execution of the CI pipeline +//! lifecycle without real IO (no SSH, no cloud API, no HTTP). +//! +//! Follows the same pattern as `crates/simulation/src/distribution/sim.rs`: +//! configure → run rounds → collect trace → analyze properties. + +use std::collections::HashMap; + +use swactor_ci::pipeline::PipelineExecution; +use swactor_ci::yaml::{self, CiYaml}; +use swactor_ci::{ + JobId, JobStatus, PipelineId, ProvisionRequest, StatusUpdate, WebhookEvent, +}; + +// ─── Simulation Config ────────────────────────────────────────────────────── + +/// Configuration for a CI simulation run. +#[derive(Debug, Clone)] +pub struct CiSimConfig { + pub name: String, + pub num_rounds: usize, + + /// CI YAML to use for all simulated repos. + pub ci_yaml: String, + + /// Webhook events to inject at specific rounds. + pub webhook_schedule: Vec<(usize, WebhookEvent)>, + + /// Rounds of latency for provisioning to complete. + pub provision_latency: usize, + + /// Probability that provisioning fails (0.0-1.0). + pub provision_failure_rate: f64, + + /// Rounds of latency for a job to complete. + pub job_duration: usize, + + /// Force specific jobs to fail: (round, job_name_substring). + pub job_failure_schedule: Vec<(usize, String)>, + + /// Rounds during which the provisioner is offline: (start_round, end_round). + pub provisioner_offline_schedule: Vec<(usize, usize)>, + + /// Round at which a specific job's instance is interrupted. + pub instance_interrupt_schedule: Vec<(usize, String)>, +} + +impl Default for CiSimConfig { + fn default() -> Self { + Self { + name: "ci-sim".into(), + num_rounds: 50, + ci_yaml: String::new(), + webhook_schedule: Vec::new(), + provision_latency: 2, + provision_failure_rate: 0.0, + job_duration: 3, + job_failure_schedule: Vec::new(), + provisioner_offline_schedule: Vec::new(), + instance_interrupt_schedule: Vec::new(), + } + } +} + +// ─── Simulation Trace ─────────────────────────────────────────────────────── + +/// Event recorded during simulation. +#[derive(Debug, Clone)] +pub enum CiSimEvent { + WebhookReceived { commit_sha: String }, + PipelineCreated { pipeline_id: PipelineId, name: String }, + ProvisionRequested { job_id: JobId }, + ProvisionCompleted { job_id: JobId, success: bool }, + JobStarted { job_id: JobId }, + JobCompleted { job_id: JobId, passed: bool }, + JobSkipped { job_id: JobId }, + InstanceTerminated { instance_id: String }, + ProvisionerWentOffline, + ProvisionerCameOnline, + StatusUpdateEmitted(StatusUpdate), +} + +/// Per-round snapshot of simulation state. +#[derive(Debug, Clone)] +pub struct CiSimSnapshot { + pub active_pipelines: usize, + pub completed_pipelines: usize, + pub active_provisions: usize, + pub active_jobs: usize, + pub provisioner_online: bool, + pub active_instances: usize, +} + +/// Complete trace output from a CI simulation. +#[derive(Debug, Clone)] +pub struct CiSimTrace { + pub name: String, + pub events: Vec<(usize, CiSimEvent)>, + pub snapshots: Vec, + pub status_updates: Vec, + pub num_rounds: usize, + /// Final state of all pipelines. + pub final_pipelines: Vec, + /// Instances that were provisioned. + pub provisioned_instances: Vec, + /// Instances that were terminated. + pub terminated_instances: Vec, +} + +// ─── Simulation State ─────────────────────────────────────────────────────── + +/// Tracks an in-flight provision request. +struct PendingProvision { + request: ProvisionRequest, + started_round: usize, +} + +/// Tracks an in-flight job execution. +struct RunningJob { + job_id: JobId, + started_round: usize, + instance_id: String, +} + +/// Run a CI simulation and return the trace. +pub fn run_simulation(config: CiSimConfig) -> CiSimTrace { + let ci_yaml: CiYaml = yaml::parse_ci_yaml(&config.ci_yaml) + .expect("CiSimConfig.ci_yaml must be valid YAML"); + + let mut events: Vec<(usize, CiSimEvent)> = Vec::new(); + let mut snapshots: Vec = Vec::new(); + + // Coordinator state (simulated directly, not as actor). + let mut pipelines: HashMap = HashMap::new(); + let mut next_pipeline_id: u64 = 1; + let mut all_status_updates: Vec = Vec::new(); + + // Provisioner state. + let mut provisioner_online = true; + let mut pending_provisions: Vec = Vec::new(); + let mut queued_provisions: Vec = Vec::new(); + let mut provisioned_instances: Vec = Vec::new(); + let mut terminated_instances: Vec = Vec::new(); + let mut active_instances: Vec = Vec::new(); + let mut next_instance_id: u64 = 1; + + // Runner state. + let mut running_jobs: Vec = Vec::new(); + + // Simple deterministic "RNG" for provision failure decisions. + let mut rng_counter: u64 = 0x853c49e6748fea9b; + let mut det_random = || -> f64 { + rng_counter = rng_counter.wrapping_mul(6364136223846793005).wrapping_add(1); + (rng_counter >> 33) as f64 / (u32::MAX as f64) + }; + + for round in 1..=config.num_rounds { + // 1. Apply provisioner online/offline schedule. + let should_be_offline = config + .provisioner_offline_schedule + .iter() + .any(|(start, end)| round >= *start && round <= *end); + + if should_be_offline && provisioner_online { + provisioner_online = false; + events.push((round, CiSimEvent::ProvisionerWentOffline)); + + // Move pending provisions to queue. + for pending in pending_provisions.drain(..) { + queued_provisions.push(pending.request); + } + } else if !should_be_offline && !provisioner_online { + provisioner_online = true; + events.push((round, CiSimEvent::ProvisionerCameOnline)); + + // Flush queued provisions. + for request in queued_provisions.drain(..) { + pending_provisions.push(PendingProvision { + request, + started_round: round, + }); + } + + // Also resubmit any WaitingForProvisioner jobs. + let mut resubmits = Vec::new(); + for pipeline in pipelines.values_mut() { + for job in pipeline.jobs.values_mut() { + if job.status == JobStatus::WaitingForProvisioner { + job.status = JobStatus::Provisioning; + resubmits.push(ProvisionRequest { + job_id: job.job_id.clone(), + instance_spec: swactor_ci::InstanceSpec { + docker_required: job.definition.docker, + ..Default::default() + }, + }); + } + } + } + for request in resubmits { + events.push((round, CiSimEvent::ProvisionRequested { job_id: request.job_id.clone() })); + pending_provisions.push(PendingProvision { + request, + started_round: round, + }); + } + } + + // 2. Inject webhook events for this round. + for (sched_round, event) in &config.webhook_schedule { + if *sched_round == round { + events.push(( + round, + CiSimEvent::WebhookReceived { + commit_sha: event.commit_sha.clone(), + }, + )); + + let matched = yaml::matching_pipelines(&ci_yaml, event); + for pipeline_name in matched { + let pipeline_id = PipelineId(next_pipeline_id); + next_pipeline_id += 1; + + let pipeline_def = &ci_yaml.pipelines[&pipeline_name]; + let job_defs: Vec<_> = pipeline_def + .jobs + .iter() + .map(|(name, def)| yaml::to_job_definition(name, def)) + .collect(); + + let pipeline = PipelineExecution::new( + pipeline_id, + pipeline_name.clone(), + event.repo_owner.clone(), + event.repo_name.clone(), + event.commit_sha.clone(), + event.branch.clone(), + job_defs, + ); + + events.push(( + round, + CiSimEvent::PipelineCreated { + pipeline_id, + name: pipeline_name.clone(), + }, + )); + + // Emit pending status. + all_status_updates.push(StatusUpdate { + repo_owner: event.repo_owner.clone(), + repo_name: event.repo_name.clone(), + commit_sha: event.commit_sha.clone(), + state: "pending".into(), + context: format!("ci/{pipeline_name}"), + description: format!("Pipeline '{pipeline_name}' is pending"), + }); + + pipelines.insert(pipeline_id, pipeline); + } + } + } + + // 3. Complete provisions that have reached latency. + let mut completed_provisions = Vec::new(); + pending_provisions.retain(|pending| { + if round - pending.started_round >= config.provision_latency { + completed_provisions.push(pending.request.clone()); + false + } else { + true + } + }); + + for request in completed_provisions { + let should_fail = det_random() < config.provision_failure_rate; + + if should_fail { + events.push(( + round, + CiSimEvent::ProvisionCompleted { + job_id: request.job_id.clone(), + success: false, + }, + )); + + if let Some(pipeline) = pipelines.get_mut(&request.job_id.pipeline_id) { + pipeline.set_job_status( + &request.job_id.job_name, + JobStatus::Failed { + reason: "provision failed".into(), + }, + ); + } + } else { + let instance_id = format!("instance-{next_instance_id}"); + next_instance_id += 1; + provisioned_instances.push(instance_id.clone()); + active_instances.push(instance_id.clone()); + + events.push(( + round, + CiSimEvent::ProvisionCompleted { + job_id: request.job_id.clone(), + success: true, + }, + )); + + // Mark job as running and record instance. + if let Some(pipeline) = pipelines.get_mut(&request.job_id.pipeline_id) { + if let Some(job) = pipeline.jobs.get_mut(&request.job_id.job_name) { + job.status = JobStatus::Running; + job.instance_id = Some(instance_id.clone()); + } + } + + events.push(( + round, + CiSimEvent::JobStarted { + job_id: request.job_id.clone(), + }, + )); + + running_jobs.push(RunningJob { + job_id: request.job_id, + started_round: round, + instance_id, + }); + } + } + + // 4. Apply instance interruptions. + for (interrupt_round, job_name_sub) in &config.instance_interrupt_schedule { + if *interrupt_round == round { + running_jobs.retain(|rj| { + if rj.job_id.job_name.contains(job_name_sub.as_str()) { + // Instance interrupted. + if let Some(pipeline) = pipelines.get_mut(&rj.job_id.pipeline_id) { + pipeline.set_job_status(&rj.job_id.job_name, JobStatus::Interrupted); + } + events.push(( + round, + CiSimEvent::JobCompleted { + job_id: rj.job_id.clone(), + passed: false, + }, + )); + // Terminate the instance. + active_instances.retain(|id| id != &rj.instance_id); + terminated_instances.push(rj.instance_id.clone()); + events.push(( + round, + CiSimEvent::InstanceTerminated { + instance_id: rj.instance_id.clone(), + }, + )); + false + } else { + true + } + }); + } + } + + // 5. Complete jobs that have reached duration. + let mut newly_completed = Vec::new(); + running_jobs.retain(|rj| { + if round - rj.started_round >= config.job_duration { + newly_completed.push((rj.job_id.clone(), rj.instance_id.clone())); + false + } else { + true + } + }); + + for (job_id, instance_id) in newly_completed { + // Check if this job should fail per the schedule. + let should_fail = config + .job_failure_schedule + .iter() + .any(|(r, name_sub)| *r <= round && job_id.job_name.contains(name_sub.as_str())); + + let passed = !should_fail; + + if passed { + if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) { + pipeline.set_job_status(&job_id.job_name, JobStatus::Passed); + } + } else { + if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) { + pipeline.set_job_status( + &job_id.job_name, + JobStatus::Failed { + reason: "command failed".into(), + }, + ); + } + } + + events.push(( + round, + CiSimEvent::JobCompleted { + job_id: job_id.clone(), + passed, + }, + )); + + // Terminate instance. + active_instances.retain(|id| id != &instance_id); + terminated_instances.push(instance_id.clone()); + events.push(( + round, + CiSimEvent::InstanceTerminated { + instance_id: instance_id.clone(), + }, + )); + } + + // 6. Advance all pipelines: schedule newly-eligible jobs. + let pipeline_ids: Vec = pipelines.keys().copied().collect(); + for pid in pipeline_ids { + let eligible = pipelines[&pid].eligible_jobs(); + for job_name in eligible { + let job_id = JobId { + pipeline_id: pid, + job_name: job_name.clone(), + }; + let docker_required = pipelines[&pid].jobs[&job_name].definition.docker; + + let request = ProvisionRequest { + job_id: job_id.clone(), + instance_spec: swactor_ci::InstanceSpec { + docker_required, + ..Default::default() + }, + }; + + if provisioner_online { + if let Some(pipeline) = pipelines.get_mut(&pid) { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.status = JobStatus::Provisioning; + } + } + events.push(( + round, + CiSimEvent::ProvisionRequested { job_id }, + )); + pending_provisions.push(PendingProvision { + request, + started_round: round, + }); + } else { + if let Some(pipeline) = pipelines.get_mut(&pid) { + if let Some(job) = pipeline.jobs.get_mut(&job_name) { + job.status = JobStatus::WaitingForProvisioner; + } + } + queued_provisions.push(request); + } + } + + // Emit final status updates for terminal pipelines. + if let Some(pipeline) = pipelines.get(&pid) { + if pipeline.status.is_terminal() { + // Check if we already emitted a terminal status for this pipeline. + let context = format!("ci/{}", pipeline.pipeline_name); + let already_emitted = all_status_updates.iter().any(|u| { + u.context == context + && u.commit_sha == pipeline.commit_sha + && (u.state == "success" || u.state == "failure" || u.state == "error") + }); + if !already_emitted { + all_status_updates.push(StatusUpdate { + repo_owner: pipeline.repo_owner.clone(), + repo_name: pipeline.repo_name.clone(), + commit_sha: pipeline.commit_sha.clone(), + state: pipeline.status.forgejo_state().into(), + context, + description: format!( + "Pipeline '{}' {}", + pipeline.pipeline_name, + pipeline.status.forgejo_state() + ), + }); + + // Emit per-job skipped events. + for (_name, job) in &pipeline.jobs { + if job.status == JobStatus::Skipped { + events.push(( + round, + CiSimEvent::JobSkipped { + job_id: job.job_id.clone(), + }, + )); + } + } + } + } + } + } + + // 7. Snapshot. + let completed_count = pipelines.values().filter(|p| p.status.is_terminal()).count(); + snapshots.push(CiSimSnapshot { + active_pipelines: pipelines.len() - completed_count, + completed_pipelines: completed_count, + active_provisions: pending_provisions.len(), + active_jobs: running_jobs.len(), + provisioner_online, + active_instances: active_instances.len(), + }); + } + + CiSimTrace { + name: config.name, + events, + snapshots, + status_updates: all_status_updates, + num_rounds: config.num_rounds, + final_pipelines: pipelines.into_values().collect(), + provisioned_instances, + terminated_instances, + } +} + +// ─── Properties ───────────────────────────────────────────────────────────── + +/// Every webhook eventually produces a terminal Forgejo status (success/failure/error). +pub fn check_all_webhooks_terminate(trace: &CiSimTrace) -> bool { + let webhook_commits: Vec<&str> = trace + .events + .iter() + .filter_map(|(_, e)| match e { + CiSimEvent::WebhookReceived { commit_sha } => Some(commit_sha.as_str()), + _ => None, + }) + .collect(); + + for sha in webhook_commits { + let has_terminal = trace.status_updates.iter().any(|u| { + u.commit_sha == sha && (u.state == "success" || u.state == "failure" || u.state == "error") + }); + if !has_terminal { + return false; + } + } + true +} + +/// Job DAG ordering is always respected: no job runs before its `needs`. +pub fn check_dag_ordering(trace: &CiSimTrace) -> bool { + // Build a map of (pipeline_id, job_name) → round when started. + let mut started: HashMap<(u64, &str), usize> = HashMap::new(); + let mut completed: HashMap<(u64, &str), usize> = HashMap::new(); + + for (round, event) in &trace.events { + match event { + CiSimEvent::JobStarted { job_id } => { + started.insert( + (job_id.pipeline_id.0, job_id.job_name.as_str()), + *round, + ); + } + CiSimEvent::JobCompleted { job_id, .. } => { + completed.insert( + (job_id.pipeline_id.0, job_id.job_name.as_str()), + *round, + ); + } + _ => {} + } + } + + // For each pipeline, check that if job B needs job A, then A completed before B started. + for pipeline in &trace.final_pipelines { + for (name, job) in &pipeline.jobs { + if let Some(&start_round) = started.get(&(pipeline.pipeline_id.0, name.as_str())) { + for dep in &job.definition.needs { + if let Some(&dep_complete_round) = + completed.get(&(pipeline.pipeline_id.0, dep.as_str())) + { + if dep_complete_round > start_round { + return false; + } + } + } + } + } + } + true +} + +/// Every provisioned instance is eventually terminated (no resource leaks). +pub fn check_no_instance_leaks(trace: &CiSimTrace) -> bool { + // Every instance that was provisioned should also be terminated. + for instance_id in &trace.provisioned_instances { + if !trace.terminated_instances.contains(instance_id) { + return false; + } + } + true +} + +/// Coordinator state is bounded: active pipeline count doesn't grow unboundedly. +pub fn check_bounded_state(trace: &CiSimTrace, max_active: usize) -> bool { + trace + .snapshots + .iter() + .all(|s| s.active_pipelines <= max_active) +} diff --git a/crates/simulation/src/lib.rs b/crates/simulation/src/lib.rs index fcb4c4d..c7712b4 100644 --- a/crates/simulation/src/lib.rs +++ b/crates/simulation/src/lib.rs @@ -9,3 +9,6 @@ pub mod gossip; #[cfg(feature = "dashboard")] pub mod dashboard; + +#[cfg(feature = "ci")] +pub mod ci; diff --git a/crates/simulation/tests/ci_properties.rs b/crates/simulation/tests/ci_properties.rs new file mode 100644 index 0000000..31d162c --- /dev/null +++ b/crates/simulation/tests/ci_properties.rs @@ -0,0 +1,275 @@ +//! Property-based tests for the CI simulation. +//! +//! These verify invariants that should hold across all possible simulation configurations. + +use swactor_ci::{EventType, WebhookEvent}; +use simulation::ci::sim::{self, CiSimConfig}; + +fn simple_yaml() -> String { + r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + fmt: + run: cargo fmt -- --check + test: + needs: [fmt] + run: cargo test +"# + .into() +} + +fn push(branch: &str, sha: &str) -> WebhookEvent { + WebhookEvent { + event_type: EventType::Push, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: branch.into(), + commit_sha: sha.into(), + tag: None, + } +} + +// ─── Property: Every webhook produces a terminal status ───────────────────── + +#[test] +fn property_all_webhooks_terminate_single() { + let config = CiSimConfig { + name: "prop-single".into(), + num_rounds: 40, + ci_yaml: simple_yaml(), + webhook_schedule: vec![(1, push("main", "sha-1"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_all_webhooks_terminate(&trace), + "single webhook must terminate" + ); +} + +#[test] +fn property_all_webhooks_terminate_burst() { + // Burst of webhooks all at once. + let config = CiSimConfig { + name: "prop-burst".into(), + num_rounds: 60, + ci_yaml: simple_yaml(), + webhook_schedule: (1..=5) + .map(|i| (1, push("main", &format!("sha-burst-{i}")))) + .collect(), + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_all_webhooks_terminate(&trace), + "burst of 5 webhooks must all terminate" + ); +} + +#[test] +fn property_all_webhooks_terminate_staggered() { + // Webhooks spread across rounds. + let config = CiSimConfig { + name: "prop-staggered".into(), + num_rounds: 60, + ci_yaml: simple_yaml(), + webhook_schedule: vec![ + (1, push("main", "sha-s1")), + (5, push("main", "sha-s2")), + (10, push("main", "sha-s3")), + (15, push("main", "sha-s4")), + ], + provision_latency: 2, + job_duration: 3, + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_all_webhooks_terminate(&trace), + "staggered webhooks must all terminate" + ); +} + +// ─── Property: DAG ordering is always respected ───────────────────────────── + +#[test] +fn property_dag_ordering_always_respected() { + // Deep chain: a → b → c → d + let yaml = r#" +pipelines: + deep: + triggers: + - event: push + branches: ["*"] + jobs: + a: + run: echo a + b: + needs: [a] + run: echo b + c: + needs: [b] + run: echo c + d: + needs: [c] + run: echo d +"#; + let config = CiSimConfig { + name: "prop-dag-deep".into(), + num_rounds: 40, + ci_yaml: yaml.into(), + webhook_schedule: vec![(1, push("main", "sha-dag"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_dag_ordering(&trace), + "deep DAG ordering must be respected" + ); +} + +#[test] +fn property_dag_ordering_diamond() { + let yaml = r#" +pipelines: + diamond: + triggers: + - event: push + branches: ["*"] + jobs: + root: + run: echo root + left: + needs: [root] + run: echo left + right: + needs: [root] + run: echo right + merge: + needs: [left, right] + run: echo merge +"#; + let config = CiSimConfig { + name: "prop-dag-diamond".into(), + num_rounds: 40, + ci_yaml: yaml.into(), + webhook_schedule: vec![(1, push("main", "sha-diamond"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_dag_ordering(&trace), + "diamond DAG ordering must be respected" + ); +} + +// ─── Property: No instance leaks ──────────────────────────────────────────── + +#[test] +fn property_no_instance_leaks_under_failures() { + let config = CiSimConfig { + name: "prop-no-leaks-fail".into(), + num_rounds: 40, + ci_yaml: simple_yaml(), + webhook_schedule: vec![ + (1, push("main", "sha-leak1")), + (3, push("main", "sha-leak2")), + ], + provision_latency: 1, + job_duration: 2, + job_failure_schedule: vec![(0, "fmt".into())], + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_no_instance_leaks(&trace), + "no instance leaks even when jobs fail" + ); +} + +#[test] +fn property_no_instance_leaks_under_interruption() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + test: + run: cargo test +"#; + let config = CiSimConfig { + name: "prop-no-leaks-interrupt".into(), + num_rounds: 40, + ci_yaml: yaml.into(), + webhook_schedule: vec![(1, push("main", "sha-int"))], + provision_latency: 1, + job_duration: 5, + instance_interrupt_schedule: vec![(4, "test".into())], + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_no_instance_leaks(&trace), + "interrupted instances must be terminated" + ); +} + +// ─── Property: Bounded state ──────────────────────────────────────────────── + +#[test] +fn property_bounded_state_under_rapid_pushes() { + let config = CiSimConfig { + name: "prop-bounded".into(), + num_rounds: 100, + ci_yaml: simple_yaml(), + webhook_schedule: (1..=20) + .map(|i| (i, push("main", &format!("sha-rapid-{i}")))) + .collect(), + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + let trace = sim::run_simulation(config); + + // With 20 pushes, each triggering 1 pipeline with 2 jobs, we should never + // have more than 20 active pipelines at once (and in practice much fewer). + assert!( + sim::check_bounded_state(&trace, 20), + "active pipelines should be bounded" + ); +} + +// ─── Property: Provisioner offline doesn't lose work ──────────────────────── + +#[test] +fn property_provisioner_offline_eventually_resolves() { + let config = CiSimConfig { + name: "prop-offline-resolve".into(), + num_rounds: 60, + ci_yaml: simple_yaml(), + webhook_schedule: vec![(3, push("main", "sha-offline"))], + provision_latency: 1, + job_duration: 2, + provisioner_offline_schedule: vec![(1, 15)], + ..Default::default() + }; + let trace = sim::run_simulation(config); + assert!( + sim::check_all_webhooks_terminate(&trace), + "webhooks during provisioner outage must still terminate" + ); +} diff --git a/crates/simulation/tests/ci_scenarios.rs b/crates/simulation/tests/ci_scenarios.rs new file mode 100644 index 0000000..33317eb --- /dev/null +++ b/crates/simulation/tests/ci_scenarios.rs @@ -0,0 +1,368 @@ +//! Scenario tests for the CI simulation. +//! +//! Each test tells a story: set up a scenario, run the simulation, verify outcomes. + +use swactor_ci::{EventType, WebhookEvent}; +use simulation::ci::sim::{self, CiSimConfig, CiSimEvent}; + +fn basic_ci_yaml() -> String { + r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + exclude: ["master"] + jobs: + fmt: + run: cargo fmt -- --check + clippy: + run: cargo clippy + test: + needs: [fmt, clippy] + run: cargo test + full: + triggers: + - event: push + branches: ["master"] + jobs: + test: + run: cargo test --all-features + timeout: 600 + bench: + needs: [test] + run: cargo bench +"# + .into() +} + +fn push_event(branch: &str, sha: &str) -> WebhookEvent { + WebhookEvent { + event_type: EventType::Push, + repo_owner: "user".into(), + repo_name: "repo".into(), + branch: branch.into(), + commit_sha: sha.into(), + tag: None, + } +} + +// ─── Scenario: Single push triggers correct pipeline ──────────────────────── + +#[test] +fn single_push_to_feature_branch_triggers_check_pipeline() { + let config = CiSimConfig { + name: "single-push-feature".into(), + num_rounds: 30, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![(1, push_event("feature-x", "abc123"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // A pipeline was created. + let pipeline_created = trace + .events + .iter() + .filter(|(_, e)| matches!(e, CiSimEvent::PipelineCreated { .. })) + .count(); + assert_eq!(pipeline_created, 1, "exactly one pipeline should be created"); + + // The pipeline name should be "check" (not "full", since branch is not master). + let check_created = trace.events.iter().any(|(_, e)| { + matches!(e, CiSimEvent::PipelineCreated { name, .. } if name == "check") + }); + assert!(check_created, "pipeline 'check' should be created"); + + // All jobs eventually complete (fmt, clippy, test). + let completed_jobs: Vec<_> = trace + .events + .iter() + .filter_map(|(_, e)| match e { + CiSimEvent::JobCompleted { job_id, passed } => Some((job_id.job_name.clone(), *passed)), + _ => None, + }) + .collect(); + + assert_eq!(completed_jobs.len(), 3, "all 3 jobs should complete"); + assert!( + completed_jobs.iter().all(|(_, passed)| *passed), + "all jobs should pass" + ); + + // A terminal Forgejo status is emitted. + assert!( + sim::check_all_webhooks_terminate(&trace), + "webhook should produce terminal status" + ); +} + +// ─── Scenario: Push to master triggers full pipeline ──────────────────────── + +#[test] +fn push_to_master_triggers_full_pipeline() { + let config = CiSimConfig { + name: "push-master".into(), + num_rounds: 30, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![(1, push_event("master", "def456"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + let full_created = trace.events.iter().any(|(_, e)| { + matches!(e, CiSimEvent::PipelineCreated { name, .. } if name == "full") + }); + assert!(full_created, "pipeline 'full' should be created"); + + // Both test and bench should eventually complete. + let completed: Vec = trace + .events + .iter() + .filter_map(|(_, e)| match e { + CiSimEvent::JobCompleted { job_id, .. } => Some(job_id.job_name.clone()), + _ => None, + }) + .collect(); + + assert!(completed.contains(&"test".to_string()), "test should complete"); + assert!(completed.contains(&"bench".to_string()), "bench should complete"); +} + +// ─── Scenario: Jobs execute in DAG order ──────────────────────────────────── + +#[test] +fn jobs_execute_in_dag_order() { + let config = CiSimConfig { + name: "dag-order".into(), + num_rounds: 30, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![(1, push_event("feature-y", "aaa111"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // test should start after fmt and clippy complete. + assert!( + sim::check_dag_ordering(&trace), + "DAG ordering must be respected" + ); +} + +// ─── Scenario: Job failure skips dependents ───────────────────────────────── + +#[test] +fn job_failure_skips_downstream_dependents() { + let config = CiSimConfig { + name: "failure-skip".into(), + num_rounds: 30, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![(1, push_event("feature-z", "bbb222"))], + provision_latency: 1, + job_duration: 2, + // fmt will fail, so test (which needs fmt) should be skipped. + job_failure_schedule: vec![(0, "fmt".into())], + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // test should be skipped. + let test_skipped = trace.events.iter().any(|(_, e)| { + matches!(e, CiSimEvent::JobSkipped { job_id } if job_id.job_name == "test") + }); + assert!(test_skipped, "test job should be skipped when fmt fails"); + + // Pipeline should be marked as failed. + let pipeline_failed = trace.status_updates.iter().any(|u| { + u.context == "ci/check" && u.state == "failure" + }); + assert!(pipeline_failed, "pipeline should report failure status"); +} + +// ─── Scenario: Parallel pushes execute independently ──────────────────────── + +#[test] +fn parallel_pushes_execute_independently() { + let config = CiSimConfig { + name: "parallel-pushes".into(), + num_rounds: 40, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![ + (1, push_event("feature-a", "ccc333")), + (1, push_event("feature-b", "ddd444")), + ], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // Two pipelines should be created. + let pipeline_count = trace + .events + .iter() + .filter(|(_, e)| matches!(e, CiSimEvent::PipelineCreated { .. })) + .count(); + assert_eq!(pipeline_count, 2, "two pipelines should be created"); + + // Both should have terminal status. + assert!( + sim::check_all_webhooks_terminate(&trace), + "both webhooks should produce terminal statuses" + ); +} + +// ─── Scenario: Provisioner goes offline, jobs queue and resume ────────────── + +#[test] +fn provisioner_offline_queues_then_resumes() { + let config = CiSimConfig { + name: "provisioner-offline".into(), + num_rounds: 50, + ci_yaml: basic_ci_yaml(), + // Push at round 3, provisioner offline rounds 1-10. + webhook_schedule: vec![(3, push_event("feature-q", "eee555"))], + provision_latency: 1, + job_duration: 2, + provisioner_offline_schedule: vec![(1, 10)], + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // Provisioner went offline and came back. + let went_offline = trace + .events + .iter() + .any(|(_, e)| matches!(e, CiSimEvent::ProvisionerWentOffline)); + let came_online = trace + .events + .iter() + .any(|(_, e)| matches!(e, CiSimEvent::ProvisionerCameOnline)); + assert!(went_offline, "provisioner should go offline"); + assert!(came_online, "provisioner should come back online"); + + // Despite the outage, all jobs should eventually complete. + assert!( + sim::check_all_webhooks_terminate(&trace), + "pipeline should complete after provisioner returns" + ); +} + +// ─── Scenario: Spot instance interrupted mid-job ──────────────────────────── + +#[test] +fn spot_instance_interruption_reports_failure() { + // Use a simpler pipeline so we have a clear target to interrupt. + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + test: + run: cargo test +"#; + let config = CiSimConfig { + name: "spot-interrupt".into(), + num_rounds: 30, + ci_yaml: yaml.into(), + webhook_schedule: vec![(1, push_event("main", "fff666"))], + provision_latency: 1, + job_duration: 5, + // Interrupt the test job at round 4 (while it's still running). + instance_interrupt_schedule: vec![(4, "test".into())], + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + // The pipeline should be terminal (failed due to interruption). + let has_failure_status = trace + .status_updates + .iter() + .any(|u| u.state == "failure" || u.state == "error"); + assert!( + has_failure_status, + "interrupted job should produce a failure status" + ); + + // The instance should be terminated. + assert!( + sim::check_no_instance_leaks(&trace), + "interrupted instance should be terminated" + ); +} + +// ─── Scenario: All jobs pass → pipeline success ───────────────────────────── + +#[test] +fn all_jobs_pass_marks_pipeline_success() { + let yaml = r#" +pipelines: + check: + triggers: + - event: push + branches: ["*"] + jobs: + lint: + run: cargo clippy + test: + needs: [lint] + run: cargo test +"#; + let config = CiSimConfig { + name: "all-pass".into(), + num_rounds: 30, + ci_yaml: yaml.into(), + webhook_schedule: vec![(1, push_event("main", "ggg777"))], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + let has_success = trace + .status_updates + .iter() + .any(|u| u.state == "success" && u.context == "ci/check"); + assert!(has_success, "pipeline should be marked as success"); +} + +// ─── Scenario: Every provisioned instance is terminated ───────────────────── + +#[test] +fn no_instance_resource_leaks() { + let config = CiSimConfig { + name: "no-leaks".into(), + num_rounds: 40, + ci_yaml: basic_ci_yaml(), + webhook_schedule: vec![ + (1, push_event("feature-1", "h1")), + (5, push_event("feature-2", "h2")), + ], + provision_latency: 1, + job_duration: 2, + ..Default::default() + }; + + let trace = sim::run_simulation(config); + + assert!( + sim::check_no_instance_leaks(&trace), + "all provisioned instances must be terminated" + ); +}