diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -1,76 +1,76 @@ -[package] -name = "syncode-control" -description = "SynCode control plane" -version.workspace = true -edition.workspace = true -rust-version.workspace = true -license.workspace = true -repository.workspace = true -publish = false - -[workspace] -members = [ - ".", - "crates/control-node", - "crates/control-nodes", - "crates/control-runs", - "crates/control-store", -] -resolver = "3" - -[workspace.package] -version = "0.5.0" -edition = "2024" -rust-version = "1.95" -license = "MIT" -repository = "https://syncode.sh/syncode/control" - -[workspace.lints.rust] -unsafe_code = "forbid" - -[workspace.lints.clippy] -expect_used = "deny" -panic = "deny" -unwrap_used = "deny" - -[dependencies] -base64 = "0.22.1" -clap = { version = "4.6.4", features = ["derive", "env"] } -flate2 = "1.1.5" -futures = "0.3.31" -hex = "0.4.3" -reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] } -serde = { version = "1.0.229", features = ["derive"] } -url = "2.5.8" -serde_json = "1.0.149" -syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } -syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } -syncode-control-node = { path = "crates/control-node" } -syncode-control-nodes = { path = "crates/control-nodes" } -syncode-control-runs = { path = "crates/control-runs" } -syncode-control-store = { version = "0.5.0", path = "crates/control-store" } -syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" } -thiserror = "2.0.19" -tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal"] } -tonic = "0.14.6" -axum = "0.8.9" -hmac = "0.13.0" -sha2 = "0.11.0" -tar = "0.4.44" -uuid = { version = "1.24.0", features = ["v4"] } - -[dev-dependencies] -base64 = "0.22.1" -hmac = "0.13.0" -reqwest = { version = "0.13.4", features = ["json", "rustls"] } -serde_json = "1.0.149" -sha2 = "0.11.0" -sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] } -syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } -syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } -tokio-stream = "0.1.18" -tempfile = "3.27.0" -tower = { version = "0.5.3", features = ["util"] } - -[lints] -workspace = true +[package] +name = "syncode-control" +description = "SynCode control plane" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +publish = false + +[workspace] +members = [ + ".", + "crates/control-node", + "crates/control-nodes", + "crates/control-runs", + "crates/control-store", +] +resolver = "3" + +[workspace.package] +version = "0.5.0" +edition = "2024" +rust-version = "1.95" +license = "MIT" +repository = "https://syncode.sh/syncode/control" + +[workspace.lints.rust] +unsafe_code = "forbid" + +[workspace.lints.clippy] +expect_used = "deny" +panic = "deny" +unwrap_used = "deny" + +[dependencies] +base64 = "0.22.1" +clap = { version = "4.6.4", features = ["derive", "env"] } +flate2 = "1.1.5" +futures = "0.3.31" +hex = "0.4.3" +reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] } +serde = { version = "1.0.229", features = ["derive"] } +url = "2.5.8" +serde_json = "1.0.149" +syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } +syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } +syncode-control-node = { path = "crates/control-node" } +syncode-control-nodes = { path = "crates/control-nodes" } +syncode-control-runs = { path = "crates/control-runs" } +syncode-control-store = { version = "0.5.0", path = "crates/control-store" } +syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" } +thiserror = "2.0.19" +tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal", "time"] } +tonic = "0.14.6" +axum = "0.8.9" +hmac = "0.13.0" +sha2 = "0.11.0" +tar = "0.4.44" +uuid = { version = "1.24.0", features = ["v4"] } + +[dev-dependencies] +base64 = "0.22.1" +hmac = "0.13.0" +reqwest = { version = "0.13.4", features = ["json", "rustls"] } +serde_json = "1.0.149" +sha2 = "0.11.0" +sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] } +syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } +syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" } +tokio-stream = "0.1.18" +tempfile = "3.27.0" +tower = { version = "0.5.3", features = ["util"] } + +[lints] +workspace = true diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -1,24 +1,27 @@ -pub mod action_delivery; -pub mod action_oci; -pub mod action_repository; -pub mod action_store; -pub mod actions; -pub mod admin; -pub mod checks; -pub mod events; -pub mod intake; -pub mod maintenance; -mod native_events; -pub mod projection; -#[path = "forge.rs"] -pub mod repository; -pub mod repository_grpc; -pub mod repository_sources; -pub mod reusable; -mod secret_reference_syntax; -pub mod secret_references; -pub mod secrets; -pub mod sources; -pub mod token; -pub mod trigger; -pub mod webhook; +pub mod action_delivery; +pub mod action_oci; +pub mod action_repository; +pub mod action_store; +pub mod actions; +pub mod actions_read; +pub mod actions_read_identity; +pub mod admin; +pub mod check_events; +pub mod checks; +pub mod events; +pub mod intake; +pub mod maintenance; +mod native_events; +pub mod projection; +#[path = "forge.rs"] +pub mod repository; +pub mod repository_grpc; +pub mod repository_sources; +pub mod reusable; +mod secret_reference_syntax; +pub mod secret_references; +pub mod secrets; +pub mod sources; +pub mod token; +pub mod trigger; +pub mod webhook; diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -1,326 +1,374 @@ -use std::error::Error; -use std::net::SocketAddr; -use std::path::PathBuf; -use std::sync::Arc; - -use clap::{Parser, Subcommand, ValueEnum}; -use syncode_control::action_delivery::router as action_delivery_router; -use syncode_control::action_oci::PinnedOciResolver; -use syncode_control::action_repository::ForgeActionRepository; -use syncode_control::action_store::FileActionStore; -use syncode_control::actions::ActionResolver; -use syncode_control::admin::{Admin, router as admin_router}; -use syncode_control::checks::Checks; -use syncode_control::maintenance; -use syncode_control::projection; -use syncode_control::repository::{RepositoryContents, RepositorySecrets}; -use syncode_control::repository_grpc::NativeRepositoryContents; -use syncode_control::repository_sources::RepositorySources; -use syncode_control::secrets::RuntimeSecrets; -use syncode_control::token::EnrolmentScope; -use syncode_control::webhook::{Intake, router}; -use syncode_control_node::{ - ArtifactTokenAuthority, CapabilityAuthority, GeneratedChecksServer, GeneratedSecretsServer, - GeneratedServer, NodeSessionServer, ProjectionClient, -}; -use syncode_control_nodes::{Nodes, Scope}; -use syncode_control_runs::{ - AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy, -}; -use syncode_control_store::Postgres; -use tokio::net::TcpListener; -use tonic::transport::Server; -use url::Url; - -#[derive(Debug, Parser)] -#[command(name = "syncode-control", version, about)] -struct Arguments { - #[command(subcommand)] - command: Option, - - /// Where nodes open their session. - #[arg(long, default_value = "127.0.0.1:8090")] - listen: SocketAddr, - - /// Where the native repository event feed delivers events. - #[arg(long, default_value = "127.0.0.1:8091")] - listen_events: SocketAddr, - - /// The repository service this control plane reads workflows from. - #[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")] - repository_source: Url, - - /// A token allowed to read repository contents. - #[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)] - repository_source_token: String, - - /// Where native repositories are read over gRPC. - #[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")] - repository_grpc: String, - - #[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")] - action_mirror_url: Url, - - #[arg( - long, - env = "SYNCODE_ACTION_ALLOWLIST", - value_delimiter = ',', - num_args = 1.. - )] - action_allowlist: Vec, - - #[arg( - long, - env = "SYNCODE_ACTION_OCI_REGISTRIES", - value_delimiter = ',', - num_args = 1.. - )] - action_oci_registries: Vec, - - #[arg(long, env = "SYNCODE_ACTION_STORE")] - action_store: PathBuf, - - #[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")] - action_artifact_public_url: Url, - - #[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")] - secret_source: Url, - - #[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)] - secret_source_token: String, - - #[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)] - capability_signing_key: String, - - #[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)] - artifact_signing_key: String, - - #[arg(long, env = "SYNCODE_PROJECTION_URL")] - projection_url: Url, - - #[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)] - projection_token: String, - - #[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")] - projection_min_run_number: u64, - - /// The secret used to sign the native event feed. - #[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)] - event_feed_secret: String, - - /// Bearer token protecting the operational API. - #[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)] - admin_token: Option, - - /// Where the run log is kept. - #[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)] - database: String, - - #[arg(long, default_value_t = 8)] - database_connections: u32, - - #[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)] - organization_concurrency: u32, - - #[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)] - repository_concurrency: u32, - - #[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)] - principal_concurrency: u32, - - #[arg( - long, - env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA", - default_value_t = 10_000 - )] - organization_queue_quota: u32, - - #[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)] - principal_queue_quota: u32, - - #[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)] - audit_retention_days: u32, - - /// Whether compiled shadow runs may be assigned to nodes. - #[arg(long, value_enum, default_value_t = Mode::Shadow)] - mode: Mode, -} - -#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] -enum Mode { - Shadow, - Active, -} - -#[derive(Debug, Subcommand)] -enum Command { - /// Issue an enrolment token a node can spend for an identity. - /// - /// The token is stored before it is printed, so one that reaches an - /// operator is one the control plane will honour. - IssueToken { - /// How far the token reaches: `instance`, `organisation:` or - /// `repository:/`. - #[arg(long, default_value = "instance")] - scope: String, - }, - /// Revoke a node identity immediately. - RevokeNode { - /// The node UUID printed at enrolment. - node: String, - }, -} - -#[tokio::main] -async fn main() -> Result<(), Box> { - let arguments = Arguments::parse(); - // The log is read before anything is served: a node that reconnects has to - // meet the run it was holding, not a control plane that has forgotten it. - let store = Postgres::connect(&arguments.database, arguments.database_connections).await?; - let policy = SchedulerPolicy::new( - arguments.organization_concurrency, - arguments.repository_concurrency, - arguments.principal_concurrency, - arguments.organization_queue_quota, - arguments.principal_queue_quota, - ); - let runs = Runs::restored_with_policy(store.clone(), policy).await?; - let nodes = Nodes::restored(store.clone()).await?; - - if let Some(command) = arguments.command { - match command { - Command::IssueToken { scope } => { - let scope: Scope = scope.parse::()?.into(); - let token = nodes.issue_token(scope).await?; - println!("{}", token.secret().expose()); - } - Command::RevokeNode { node } => { - nodes.revoke(node.parse::()?).await?; - } - } - return Ok(()); - } - - store - .record_audit(AuditEvent::new("control.configuration", "applied").after( - AuditPayload::Configuration(AuditConfiguration { - mode: match arguments.mode { - Mode::Shadow => AuditControlMode::Shadow, - Mode::Active => AuditControlMode::Active, - }, - organization_concurrency: arguments.organization_concurrency, - repository_concurrency: arguments.repository_concurrency, - principal_concurrency: arguments.principal_concurrency, - organization_queue_quota: arguments.organization_queue_quota, - principal_queue_quota: arguments.principal_queue_quota, - audit_retention_days: arguments.audit_retention_days, - projection_min_run_number: arguments.projection_min_run_number, - }), - )) - .await?; - tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone())); - tokio::spawn(maintenance::retain_audit( - store.clone(), - arguments.audit_retention_days, - )); - - let action_mirror = arguments - .action_mirror_url - .to_string() - .parse::()?; - let action_allowlist = arguments - .action_allowlist - .into_iter() - .map(|repository| { - repository - .to_string() - .parse::() - }) - .collect::, _>>()?; - let action_store = FileActionStore::new(arguments.action_store); - let action_resolver = ActionResolver::new( - ForgeActionRepository::new(arguments.repository_source_token.clone()), - action_store.clone(), - PinnedOciResolver::new(arguments.action_oci_registries)?, - action_mirror, - action_allowlist, - ); - let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).await?; - let repository_sources = RepositorySources::new( - native_repository, - RepositoryContents::new( - arguments.repository_source, - arguments.repository_source_token, - ), - ); - let intake = Arc::new(Intake::new( - repository_sources, - runs.clone(), - action_resolver, - arguments.event_feed_secret, - )); - let admin_token = arguments - .admin_token - .filter(|token| !token.is_empty()) - .ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?; - let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?; - let http = router(intake) - .merge(admin_router(Arc::new( - Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store), - ))) - .merge(action_delivery_router( - action_store, - artifact_authority.clone(), - runs.clone(), - )); - let events = TcpListener::bind(arguments.listen_events).await?; - - let authority = CapabilityAuthority::new(arguments.capability_signing_key)?; - let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?; - let secret_service = RuntimeSecrets::new( - runs.clone(), - nodes.clone(), - RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token), - authority.clone(), - ); - - // Both ends of the control plane run for as long as the other does: without - // events there is nothing to assign, and without sessions there is nobody to - // assign it to. Whichever stops first takes the process down with it. - let node_service = match arguments.mode { - Mode::Shadow => NodeSessionServer::shadow( - runs.clone(), - nodes, - authority, - artifact_authority, - arguments.action_artifact_public_url, - projection.clone(), - ), - Mode::Active => NodeSessionServer::new( - runs.clone(), - nodes, - authority, - artifact_authority, - arguments.action_artifact_public_url, - projection.clone(), - ), - }; - tokio::select! { - served = axum::serve(events, http).into_future() => served?, - served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?, - served = Server::builder() - .add_service(GeneratedServer::new(node_service)) - .add_service(GeneratedSecretsServer::new(secret_service)) - .add_service(GeneratedChecksServer::new(Checks::new(runs.clone()))) - .serve_with_shutdown(arguments.listen, shutdown()) => served?, - } - - Ok(()) -} - -async fn shutdown() { - if let Err(error) = tokio::signal::ctrl_c().await { - eprintln!("cannot listen for shutdown, keeping the service running: {error}"); - std::future::pending::<()>().await; - } -} +use std::error::Error; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; + +use clap::{Parser, Subcommand, ValueEnum}; +use syncode_control::action_delivery::router as action_delivery_router; +use syncode_control::action_oci::PinnedOciResolver; +use syncode_control::action_repository::ForgeActionRepository; +use syncode_control::action_store::FileActionStore; +use syncode_control::actions::ActionResolver; +use syncode_control::actions_read::ActionsRead; +use syncode_control::actions_read_identity::IdentityActionsAuthorization; +use syncode_control::admin::{Admin, router as admin_router}; +use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig}; +use syncode_control::checks::Checks; +use syncode_control::maintenance; +use syncode_control::projection; +use syncode_control::repository::{RepositoryContents, RepositorySecrets}; +use syncode_control::repository_grpc::NativeRepositoryContents; +use syncode_control::repository_sources::RepositorySources; +use syncode_control::secrets::RuntimeSecrets; +use syncode_control::token::EnrolmentScope; +use syncode_control::webhook::{Intake, router}; +use syncode_control_node::{ + ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer, + GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient, +}; +use syncode_control_nodes::{Nodes, Scope}; +use syncode_control_runs::{ + AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy, +}; +use syncode_control_store::Postgres; +use tokio::net::TcpListener; +use tonic::transport::Server; +use url::Url; + +#[derive(Debug, Parser)] +#[command(name = "syncode-control", version, about)] +struct Arguments { + #[command(subcommand)] + command: Option, + + /// Where nodes open their session. + #[arg(long, default_value = "127.0.0.1:8090")] + listen: SocketAddr, + + /// Where the native repository event feed delivers events. + #[arg(long, default_value = "127.0.0.1:8091")] + listen_events: SocketAddr, + + /// The repository service this control plane reads workflows from. + #[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")] + repository_source: Url, + + /// A token allowed to read repository contents. + #[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)] + repository_source_token: String, + + /// Where native repositories are read over gRPC. + #[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")] + repository_grpc: String, + + #[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")] + identity_grpc: String, + + #[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)] + identity_shared_secret: String, + + #[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")] + action_mirror_url: Url, + + #[arg( + long, + env = "SYNCODE_ACTION_ALLOWLIST", + value_delimiter = ',', + num_args = 1.. + )] + action_allowlist: Vec, + + #[arg( + long, + env = "SYNCODE_ACTION_OCI_REGISTRIES", + value_delimiter = ',', + num_args = 1.. + )] + action_oci_registries: Vec, + + #[arg(long, env = "SYNCODE_ACTION_STORE")] + action_store: PathBuf, + + #[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")] + action_artifact_public_url: Url, + + #[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")] + secret_source: Url, + + #[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)] + secret_source_token: String, + + #[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)] + capability_signing_key: String, + + #[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)] + artifact_signing_key: String, + + #[arg(long, env = "SYNCODE_PROJECTION_URL")] + projection_url: Url, + + #[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)] + projection_token: String, + + #[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")] + projection_min_run_number: u64, + + /// The secret used to sign the native event feed. + #[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)] + event_feed_secret: String, + + #[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")] + collaboration_event_url: Url, + + #[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)] + collaboration_event_secret: String, + + #[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")] + source_node_id: uuid::Uuid, + + #[arg( + long, + env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS", + default_value_t = 1 + )] + check_event_interval_seconds: u64, + + #[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)] + check_event_batch: i64, + + /// Bearer token protecting the operational API. + #[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)] + admin_token: Option, + + /// Where the run log is kept. + #[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)] + database: String, + + #[arg(long, default_value_t = 8)] + database_connections: u32, + + #[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)] + organization_concurrency: u32, + + #[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)] + repository_concurrency: u32, + + #[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)] + principal_concurrency: u32, + + #[arg( + long, + env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA", + default_value_t = 10_000 + )] + organization_queue_quota: u32, + + #[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)] + principal_queue_quota: u32, + + #[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)] + audit_retention_days: u32, + + /// Whether compiled shadow runs may be assigned to nodes. + #[arg(long, value_enum, default_value_t = Mode::Shadow)] + mode: Mode, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +enum Mode { + Shadow, + Active, +} + +#[derive(Debug, Subcommand)] +enum Command { + /// Issue an enrolment token a node can spend for an identity. + /// + /// The token is stored before it is printed, so one that reaches an + /// operator is one the control plane will honour. + IssueToken { + /// How far the token reaches: `instance`, `organisation:` or + /// `repository:/`. + #[arg(long, default_value = "instance")] + scope: String, + }, + /// Revoke a node identity immediately. + RevokeNode { + /// The node UUID printed at enrolment. + node: String, + }, +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let arguments = Arguments::parse(); + // The log is read before anything is served: a node that reconnects has to + // meet the run it was holding, not a control plane that has forgotten it. + let store = Postgres::connect(&arguments.database, arguments.database_connections).await?; + let policy = SchedulerPolicy::new( + arguments.organization_concurrency, + arguments.repository_concurrency, + arguments.principal_concurrency, + arguments.organization_queue_quota, + arguments.principal_queue_quota, + ); + let runs = Runs::restored_with_policy(store.clone(), policy).await?; + let nodes = Nodes::restored(store.clone()).await?; + + if let Some(command) = arguments.command { + match command { + Command::IssueToken { scope } => { + let scope: Scope = scope.parse::()?.into(); + let token = nodes.issue_token(scope).await?; + println!("{}", token.secret().expose()); + } + Command::RevokeNode { node } => { + nodes.revoke(node.parse::()?).await?; + } + } + return Ok(()); + } + + store + .record_audit(AuditEvent::new("control.configuration", "applied").after( + AuditPayload::Configuration(AuditConfiguration { + mode: match arguments.mode { + Mode::Shadow => AuditControlMode::Shadow, + Mode::Active => AuditControlMode::Active, + }, + organization_concurrency: arguments.organization_concurrency, + repository_concurrency: arguments.repository_concurrency, + principal_concurrency: arguments.principal_concurrency, + organization_queue_quota: arguments.organization_queue_quota, + principal_queue_quota: arguments.principal_queue_quota, + audit_retention_days: arguments.audit_retention_days, + projection_min_run_number: arguments.projection_min_run_number, + }), + )) + .await?; + tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone())); + tokio::spawn(maintenance::retain_audit( + store.clone(), + arguments.audit_retention_days, + )); + + let action_mirror = arguments + .action_mirror_url + .to_string() + .parse::()?; + let action_allowlist = arguments + .action_allowlist + .into_iter() + .map(|repository| { + repository + .to_string() + .parse::() + }) + .collect::, _>>()?; + let action_store = FileActionStore::new(arguments.action_store); + let action_resolver = ActionResolver::new( + ForgeActionRepository::new(arguments.repository_source_token.clone()), + action_store.clone(), + PinnedOciResolver::new(arguments.action_oci_registries)?, + action_mirror, + action_allowlist, + ); + let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).await?; + let repository_sources = RepositorySources::new( + native_repository, + RepositoryContents::new( + arguments.repository_source, + arguments.repository_source_token, + ), + ); + let intake = Arc::new(Intake::new( + repository_sources, + runs.clone(), + action_resolver, + arguments.event_feed_secret, + )); + let admin_token = arguments + .admin_token + .filter(|token| !token.is_empty()) + .ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?; + let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?; + let http = router(intake) + .merge(admin_router(Arc::new( + Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()), + ))) + .merge(action_delivery_router( + action_store, + artifact_authority.clone(), + runs.clone(), + )); + let events = TcpListener::bind(arguments.listen_events).await?; + let check_events = CheckEventPublisher::new( + store.clone(), + CheckEventPublisherConfig { + endpoint: arguments.collaboration_event_url.as_str(), + secret: arguments.collaboration_event_secret, + source_node_id: arguments.source_node_id, + interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds), + batch: arguments.check_event_batch, + }, + )?; + + let authority = CapabilityAuthority::new(arguments.capability_signing_key)?; + let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?; + let secret_service = RuntimeSecrets::new( + runs.clone(), + nodes.clone(), + RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token), + authority.clone(), + ); + let actions_read = ActionsRead::new( + runs.clone(), + IdentityActionsAuthorization::connect( + arguments.identity_grpc, + arguments.identity_shared_secret, + ) + .await?, + ); + + // Both ends of the control plane run for as long as the other does: without + // events there is nothing to assign, and without sessions there is nobody to + // assign it to. Whichever stops first takes the process down with it. + let node_service = match arguments.mode { + Mode::Shadow => NodeSessionServer::shadow( + runs.clone(), + nodes, + authority, + artifact_authority, + arguments.action_artifact_public_url, + projection.clone(), + ), + Mode::Active => NodeSessionServer::new( + runs.clone(), + nodes, + authority, + artifact_authority, + arguments.action_artifact_public_url, + projection.clone(), + ), + }; + tokio::select! { + served = axum::serve(events, http).into_future() => served?, + served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?, + published = check_events.run() => published?, + served = Server::builder() + .add_service(GeneratedServer::new(node_service)) + .add_service(GeneratedSecretsServer::new(secret_service)) + .add_service(GeneratedChecksServer::new(Checks::new(runs.clone()))) + .add_service(GeneratedActionsReadServer::new(actions_read)) + .serve_with_shutdown(arguments.listen, shutdown()) => served?, + } + + Ok(()) +} + +async fn shutdown() { + if let Err(error) = tokio::signal::ctrl_c().await { + eprintln!("cannot listen for shutdown, keeping the service running: {error}"); + std::future::pending::<()>().await; + } +} diff --git a/crates/control-node/build.rs b/crates/control-node/build.rs --- a/crates/control-node/build.rs +++ b/crates/control-node/build.rs @@ -1,15 +1,19 @@ -use std::error::Error; - -fn main() -> Result<(), Box> { - let mut config = prost_build::Config::new(); - config.protoc_executable(protoc_bin_vendored::protoc_bin_path()?); - - tonic_prost_build::configure().compile_with_config( - config, - &["proto/node.proto"], - &["proto"], - )?; - - println!("cargo:rerun-if-changed=proto"); - Ok(()) -} +use std::error::Error; + +fn main() -> Result<(), Box> { + let mut config = prost_build::Config::new(); + config.protoc_executable(protoc_bin_vendored::protoc_bin_path()?); + + tonic_prost_build::configure().compile_with_config( + config, + &[ + "proto/node.proto", + "proto/actions.proto", + "proto/identity.proto", + ], + &["proto"], + )?; + + println!("cargo:rerun-if-changed=proto"); + Ok(()) +} diff --git a/crates/control-node/src/lib.rs b/crates/control-node/src/lib.rs --- a/crates/control-node/src/lib.rs +++ b/crates/control-node/src/lib.rs @@ -1,27 +1,41 @@ -mod artifact; -mod capability; -mod declaration; -mod error; -mod identity; -mod outbound; -mod projection; -mod reports; -mod session; - -pub mod wire { - //! Generated from `proto/node.proto`. - #![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)] - - include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs")); -} - -pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError}; -pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError}; -pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError}; -pub use session::NodeSessionServer; -pub use wire::checks_server::Checks as ChecksService; -pub use wire::checks_server::ChecksServer as GeneratedChecksServer; -pub use wire::node_session_client::NodeSessionClient; -pub use wire::node_session_server::NodeSessionServer as GeneratedServer; -pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService; -pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer; +mod artifact; +mod capability; +mod declaration; +mod error; +mod identity; +mod outbound; +mod projection; +mod reports; +mod session; + +pub mod wire { + //! Generated from `proto/node.proto`. + #![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)] + + include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs")); +} + +pub mod identity_wire { + #![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)] + + include!(concat!(env!("OUT_DIR"), "/syncode.identity.v1.rs")); +} + +pub mod actions_wire { + #![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)] + + include!(concat!(env!("OUT_DIR"), "/syncode.control.v1.rs")); +} + +pub use actions_wire::actions_read_server::ActionsRead as ActionsReadService; +pub use actions_wire::actions_read_server::ActionsReadServer as GeneratedActionsReadServer; +pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError}; +pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError}; +pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError}; +pub use session::NodeSessionServer; +pub use wire::checks_server::Checks as ChecksService; +pub use wire::checks_server::ChecksServer as GeneratedChecksServer; +pub use wire::node_session_client::NodeSessionClient; +pub use wire::node_session_server::NodeSessionServer as GeneratedServer; +pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService; +pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer; diff --git a/crates/control-runs/src/lib.rs b/crates/control-runs/src/lib.rs --- a/crates/control-runs/src/lib.rs +++ b/crates/control-runs/src/lib.rs @@ -1,35 +1,37 @@ -mod assignment; -mod audit; -mod diagnosis; -mod identity; -mod job; -mod lease; -mod log; -mod origin; -mod output; -mod projection; -mod record; -mod registry; -mod run; -mod scheduler; - -pub use assignment::{Assignment, Dependency}; -pub use audit::{ - AuditConfiguration, AuditContainerRuntime, AuditControlMode, AuditEnrolment, AuditEvent, - AuditJob, AuditJobs, AuditNodeCapabilities, AuditNodeCapacity, AuditNodeLifecycle, - AuditNodeScope, AuditNodeSnapshot, AuditPayload, AuditRunAdmission, AuditVolume, -}; -pub use diagnosis::QueueDiagnosis; -pub use identity::{IdentityError, JobId, NodeId, RunId, Sequence}; -pub use job::{JobState, MatrixPolicy, QueuedJob}; -pub use lease::{Fence, Lease}; -pub use log::{Forgotten, ForgottenError, OpenedRun, RunLog, StoredRun}; -pub use origin::{Origin, RunNumber}; -pub use projection::{ProjectedJob, ProjectedJobState, ProjectedRun}; -pub use record::RunRecord; -pub use registry::{Cancellation, CancellationDispatch, Dispatch, LEASE_TERM, Runs, RunsError}; -pub use run::{Conclusion, Run, RunCommand, RunError, RunEvent, RunState, SecretReadOutcome}; -pub use scheduler::{ - Architecture, Candidate, OperatingSystem, Priority, Refusal as SchedulingRefusal, Requirements, - SchedulerPolicy, Scope as SchedulingScope, -}; +mod assignment; +mod audit; +mod diagnosis; +mod identity; +mod job; +mod lease; +mod log; +mod origin; +mod output; +mod projection; +mod read; +mod record; +mod registry; +mod run; +mod scheduler; + +pub use assignment::{Assignment, Dependency}; +pub use audit::{ + AuditConfiguration, AuditContainerRuntime, AuditControlMode, AuditEnrolment, AuditEvent, + AuditJob, AuditJobs, AuditNodeCapabilities, AuditNodeCapacity, AuditNodeLifecycle, + AuditNodeScope, AuditNodeSnapshot, AuditPayload, AuditRunAdmission, AuditVolume, +}; +pub use diagnosis::QueueDiagnosis; +pub use identity::{IdentityError, JobId, NodeId, RunId, Sequence}; +pub use job::{JobState, MatrixPolicy, QueuedJob}; +pub use lease::{Fence, Lease}; +pub use log::{Forgotten, ForgottenError, OpenedRun, RunLog, StoredRun}; +pub use origin::{Origin, RunNumber}; +pub use projection::{ProjectedJob, ProjectedJobState, ProjectedRun}; +pub use read::{JobView, RunView}; +pub use record::RunRecord; +pub use registry::{Cancellation, CancellationDispatch, Dispatch, LEASE_TERM, Runs, RunsError}; +pub use run::{Conclusion, Run, RunCommand, RunError, RunEvent, RunState, SecretReadOutcome}; +pub use scheduler::{ + Architecture, Candidate, OperatingSystem, Priority, Refusal as SchedulingRefusal, Requirements, + SchedulerPolicy, Scope as SchedulingScope, +}; diff --git a/crates/control-store/src/lib.rs b/crates/control-store/src/lib.rs --- a/crates/control-store/src/lib.rs +++ b/crates/control-store/src/lib.rs @@ -1,197 +1,214 @@ -//! What the control plane writes down, kept in Postgres. -//! -//! Nothing here decides anything. The aggregate decides, this writes the -//! decision down and reads it back, which is why a run has no stored state -//! column: its state is what replaying its events produces. A node is the other -//! shape — a current fact rather than a history — so its state is stored and -//! overwritten. - -mod audit; -mod nodes; -mod runs; - -pub use audit::RecordedAuditEvent; - -use sqlx::Executor; -use sqlx::postgres::{PgPool, PgPoolOptions}; -use thiserror::Error; - -#[derive(Debug, Error)] -pub enum StoreError { - #[error("the store cannot be reached: {0}")] - Unreachable(#[from] sqlx::Error), - - #[error("something stored cannot be read back: {0}")] - Unreadable(#[from] serde_json::Error), - - #[error("sequence {0} in the log does not fit the range a run can hold")] - Sequence(i64), - - #[error("a stored run contains no jobs")] - EmptyRun, - - #[error("audit retention of {0} days is outside the database range")] - Retention(u32), -} - -#[derive(Clone)] -pub struct Postgres { - pool: PgPool, -} - -/// The tables the log needs, created if this is the first time it runs. -/// -/// A run's events are keyed by the run and its sequence, so the same event -/// cannot be written twice and a gap cannot be filled in later by accident. -/// -/// The columns a table created by an earlier version does not have are added -/// nullable first, because there is no honest value to give a run that was -/// opened before the control plane wrote down where a run came from. -/// -/// Such a run is then dropped rather than kept, and the columns are made -/// mandatory so no later one can look like it. Inventing an origin is still off -/// the table; what changed is the cost of refusing. A run with no origin can -/// never be assigned — a node cannot be told what it is building — so keeping it -/// bought nothing, while reading it back took the whole control plane down on -/// every start, which is how one unusable row became an outage. -/// -/// An origin that cannot name its workflow file goes the same way, and for the -/// same reason: it is a shape this control plane no longer writes and can no -/// longer read, so leaving it in the table is leaving a start to fail on. -const SCHEMA: &str = r#" -CREATE TABLE IF NOT EXISTS runs ( - id UUID PRIMARY KEY, - job UUID NOT NULL, - number BIGINT NOT NULL, - origin JSONB NOT NULL, - plan BYTEA NOT NULL, - opened_at TIMESTAMPTZ NOT NULL DEFAULT now() -); -ALTER TABLE runs ADD COLUMN IF NOT EXISTS number BIGINT; -ALTER TABLE runs ADD COLUMN IF NOT EXISTS origin JSONB; -DELETE FROM runs WHERE number IS NULL OR origin IS NULL OR origin->>'workflow' IS NULL; -ALTER TABLE runs ALTER COLUMN number SET NOT NULL; -ALTER TABLE runs ALTER COLUMN origin SET NOT NULL; -CREATE TABLE IF NOT EXISTS run_jobs ( - run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, - id UUID NOT NULL, - position BIGINT NOT NULL, - key TEXT NOT NULL, - needs JSONB NOT NULL, - matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb, - priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb, - requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb, - secrets JSONB NOT NULL DEFAULT '[]'::jsonb, - plan BYTEA NOT NULL, - PRIMARY KEY (run, id), - UNIQUE (run, position) -); -ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb; -ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb; -ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb; -ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS secrets JSONB NOT NULL DEFAULT '[]'::jsonb; -INSERT INTO run_jobs (run, id, position, key, needs, plan) -SELECT id, job, 0, job::text, '[]'::jsonb, plan FROM runs -ON CONFLICT (run, id) DO NOTHING; -CREATE TABLE IF NOT EXISTS run_events ( - run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, - sequence BIGINT NOT NULL, - event JSONB NOT NULL, - recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), - PRIMARY KEY (run, sequence) -); -UPDATE run_events e SET event = to_jsonb('Opened'::text) -FROM runs r WHERE e.run = r.id AND e.event = to_jsonb('Queued'::text); -UPDATE run_events e SET event = jsonb_build_object( - 'Assigned', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Assigned')) -FROM runs r WHERE e.run = r.id AND e.event ? 'Assigned' - AND NOT ((e.event->'Assigned') ? 'job'); -UPDATE run_events e SET event = jsonb_build_object( - 'Renewed', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Renewed')) -FROM runs r WHERE e.run = r.id AND e.event ? 'Renewed' - AND NOT ((e.event->'Renewed') ? 'job'); -UPDATE run_events e SET event = jsonb_build_object( - 'Started', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Started')) -FROM runs r WHERE e.run = r.id AND e.event ? 'Started' - AND NOT ((e.event->'Started') ? 'job'); -UPDATE run_events e SET event = jsonb_build_object( - 'Finished', jsonb_build_object('job', to_jsonb(r.job), 'conclusion', e.event->'Finished')) -FROM runs r WHERE e.run = r.id AND e.event ? 'Finished' - AND NOT ((e.event->'Finished') ? 'job'); -UPDATE run_events e SET event = jsonb_build_object( - 'LeaseExpired', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'LeaseExpired')) -FROM runs r WHERE e.run = r.id AND e.event ? 'LeaseExpired' - AND NOT ((e.event->'LeaseExpired') ? 'job'); -CREATE TABLE IF NOT EXISTS run_logs ( - run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, - job UUID NOT NULL, - at_offset BIGINT NOT NULL, - lines JSONB NOT NULL, - received_at TIMESTAMPTZ NOT NULL DEFAULT now(), - PRIMARY KEY (run, job, at_offset) -); -CREATE TABLE IF NOT EXISTS nodes ( - id UUID PRIMARY KEY, - state JSONB NOT NULL, - saved_at TIMESTAMPTZ NOT NULL DEFAULT now() -); -CREATE TABLE IF NOT EXISTS enrolment_tokens ( - secret TEXT PRIMARY KEY, - token JSONB NOT NULL, - issued_at TIMESTAMPTZ NOT NULL DEFAULT now() -); -CREATE TABLE IF NOT EXISTS node_message_inbox ( - node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE, - message_id TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - payload BYTEA NOT NULL, - applied BOOLEAN NOT NULL DEFAULT false, - received_at TIMESTAMPTZ NOT NULL DEFAULT now(), - PRIMARY KEY (node, message_id), - UNIQUE (node, idempotency_key) -); -CREATE TABLE IF NOT EXISTS node_control_outbox ( - node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE, - message_id TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - payload BYTEA NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - acknowledged_at TIMESTAMPTZ, - PRIMARY KEY (node, message_id), - UNIQUE (node, idempotency_key) -); -CREATE TABLE IF NOT EXISTS audit_entries ( - id BIGSERIAL PRIMARY KEY, - occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(), - kind TEXT NOT NULL, - outcome TEXT NOT NULL, - principal TEXT, - node UUID, - run UUID, - job UUID, - correlation TEXT, - before_state JSONB, - after_state JSONB, - reason TEXT -); -CREATE INDEX IF NOT EXISTS audit_entries_occurred_at ON audit_entries (occurred_at); -CREATE INDEX IF NOT EXISTS audit_entries_run_job ON audit_entries (run, job, id); -CREATE INDEX IF NOT EXISTS audit_entries_node ON audit_entries (node, id); -"#; - -impl Postgres { - /// Connect and make sure the schema is there. - pub async fn connect(url: &str, connections: u32) -> Result { - let pool = PgPoolOptions::new() - .max_connections(connections) - .connect(url) - .await?; - pool.execute(SCHEMA).await?; - Ok(Self { pool }) - } - - #[must_use] - pub const fn pool(&self) -> &PgPool { - &self.pool - } -} +//! What the control plane writes down, kept in Postgres. +//! +//! Nothing here decides anything. The aggregate decides, this writes the +//! decision down and reads it back, which is why a run has no stored state +//! column: its state is what replaying its events produces. A node is the other +//! shape — a current fact rather than a history — so its state is stored and +//! overwritten. + +mod audit; +mod check_events; +mod nodes; +mod runs; + +pub use audit::RecordedAuditEvent; +pub use check_events::PendingCheckEvent; + +use sqlx::Executor; +use sqlx::postgres::{PgPool, PgPoolOptions}; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum StoreError { + #[error("the store cannot be reached: {0}")] + Unreachable(#[from] sqlx::Error), + + #[error("something stored cannot be read back: {0}")] + Unreadable(#[from] serde_json::Error), + + #[error("sequence {0} in the log does not fit the range a run can hold")] + Sequence(i64), + + #[error("a stored run contains no jobs")] + EmptyRun, + + #[error("audit retention of {0} days is outside the database range")] + Retention(u32), + + #[error("a stored run is invalid: {0}")] + InvalidRun(String), +} + +#[derive(Clone)] +pub struct Postgres { + pool: PgPool, +} + +/// The tables the log needs, created if this is the first time it runs. +/// +/// A run's events are keyed by the run and its sequence, so the same event +/// cannot be written twice and a gap cannot be filled in later by accident. +/// +/// The columns a table created by an earlier version does not have are added +/// nullable first, because there is no honest value to give a run that was +/// opened before the control plane wrote down where a run came from. +/// +/// Such a run is then dropped rather than kept, and the columns are made +/// mandatory so no later one can look like it. Inventing an origin is still off +/// the table; what changed is the cost of refusing. A run with no origin can +/// never be assigned — a node cannot be told what it is building — so keeping it +/// bought nothing, while reading it back took the whole control plane down on +/// every start, which is how one unusable row became an outage. +/// +/// An origin that cannot name its workflow file goes the same way, and for the +/// same reason: it is a shape this control plane no longer writes and can no +/// longer read, so leaving it in the table is leaving a start to fail on. +const SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS runs ( + id UUID PRIMARY KEY, + job UUID NOT NULL, + number BIGINT NOT NULL, + origin JSONB NOT NULL, + plan BYTEA NOT NULL, + opened_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +ALTER TABLE runs ADD COLUMN IF NOT EXISTS number BIGINT; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS origin JSONB; +DELETE FROM runs WHERE number IS NULL OR origin IS NULL OR origin->>'workflow' IS NULL; +ALTER TABLE runs ALTER COLUMN number SET NOT NULL; +ALTER TABLE runs ALTER COLUMN origin SET NOT NULL; +CREATE TABLE IF NOT EXISTS run_jobs ( + run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + id UUID NOT NULL, + position BIGINT NOT NULL, + key TEXT NOT NULL, + needs JSONB NOT NULL, + matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb, + priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb, + requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb, + secrets JSONB NOT NULL DEFAULT '[]'::jsonb, + plan BYTEA NOT NULL, + PRIMARY KEY (run, id), + UNIQUE (run, position) +); +ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb; +ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb; +ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb; +ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS secrets JSONB NOT NULL DEFAULT '[]'::jsonb; +INSERT INTO run_jobs (run, id, position, key, needs, plan) +SELECT id, job, 0, job::text, '[]'::jsonb, plan FROM runs +ON CONFLICT (run, id) DO NOTHING; +CREATE TABLE IF NOT EXISTS run_events ( + run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + sequence BIGINT NOT NULL, + event JSONB NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (run, sequence) +); +UPDATE run_events e SET event = to_jsonb('Opened'::text) +FROM runs r WHERE e.run = r.id AND e.event = to_jsonb('Queued'::text); +UPDATE run_events e SET event = jsonb_build_object( + 'Assigned', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Assigned')) +FROM runs r WHERE e.run = r.id AND e.event ? 'Assigned' + AND NOT ((e.event->'Assigned') ? 'job'); +UPDATE run_events e SET event = jsonb_build_object( + 'Renewed', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Renewed')) +FROM runs r WHERE e.run = r.id AND e.event ? 'Renewed' + AND NOT ((e.event->'Renewed') ? 'job'); +UPDATE run_events e SET event = jsonb_build_object( + 'Started', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Started')) +FROM runs r WHERE e.run = r.id AND e.event ? 'Started' + AND NOT ((e.event->'Started') ? 'job'); +UPDATE run_events e SET event = jsonb_build_object( + 'Finished', jsonb_build_object('job', to_jsonb(r.job), 'conclusion', e.event->'Finished')) +FROM runs r WHERE e.run = r.id AND e.event ? 'Finished' + AND NOT ((e.event->'Finished') ? 'job'); +UPDATE run_events e SET event = jsonb_build_object( + 'LeaseExpired', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'LeaseExpired')) +FROM runs r WHERE e.run = r.id AND e.event ? 'LeaseExpired' + AND NOT ((e.event->'LeaseExpired') ? 'job'); +CREATE TABLE IF NOT EXISTS run_logs ( + run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + job UUID NOT NULL, + at_offset BIGINT NOT NULL, + lines JSONB NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (run, job, at_offset) +); +CREATE TABLE IF NOT EXISTS check_event_outbox ( + id BIGSERIAL PRIMARY KEY, + message_id UUID NOT NULL UNIQUE, + repository_id UUID NOT NULL, + payload JSONB NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + delivered_at TIMESTAMPTZ, + last_error TEXT +); +CREATE INDEX IF NOT EXISTS check_event_outbox_pending +ON check_event_outbox (available_at, id) WHERE delivered_at IS NULL; +CREATE TABLE IF NOT EXISTS nodes ( + id UUID PRIMARY KEY, + state JSONB NOT NULL, + saved_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS enrolment_tokens ( + secret TEXT PRIMARY KEY, + token JSONB NOT NULL, + issued_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS node_message_inbox ( + node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE, + message_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + payload BYTEA NOT NULL, + applied BOOLEAN NOT NULL DEFAULT false, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (node, message_id), + UNIQUE (node, idempotency_key) +); +CREATE TABLE IF NOT EXISTS node_control_outbox ( + node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE, + message_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + payload BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + acknowledged_at TIMESTAMPTZ, + PRIMARY KEY (node, message_id), + UNIQUE (node, idempotency_key) +); +CREATE TABLE IF NOT EXISTS audit_entries ( + id BIGSERIAL PRIMARY KEY, + occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(), + kind TEXT NOT NULL, + outcome TEXT NOT NULL, + principal TEXT, + node UUID, + run UUID, + job UUID, + correlation TEXT, + before_state JSONB, + after_state JSONB, + reason TEXT +); +CREATE INDEX IF NOT EXISTS audit_entries_occurred_at ON audit_entries (occurred_at); +CREATE INDEX IF NOT EXISTS audit_entries_run_job ON audit_entries (run, job, id); +CREATE INDEX IF NOT EXISTS audit_entries_node ON audit_entries (node, id); +"#; + +impl Postgres { + /// Connect and make sure the schema is there. + pub async fn connect(url: &str, connections: u32) -> Result { + let pool = PgPoolOptions::new() + .max_connections(connections) + .connect(url) + .await?; + pool.execute(SCHEMA).await?; + Ok(Self { pool }) + } + + #[must_use] + pub const fn pool(&self) -> &PgPool { + &self.pool + } +} diff --git a/crates/control-store/src/runs.rs b/crates/control-store/src/runs.rs --- a/crates/control-store/src/runs.rs +++ b/crates/control-store/src/runs.rs @@ -1,234 +1,249 @@ -//! How a run's history is written down and read back. - -mod audit; -mod restore; - -use std::collections::BTreeMap; - -use sqlx::types::Json; -use syncode_control_runs::{ - AuditEvent, AuditJob, AuditJobs, AuditPayload, JobId, OpenedRun, Origin, Priority, QueuedJob, - Requirements, RunId, RunLog, RunRecord, StoredRun, -}; -use uuid::Uuid; - -use self::audit::run_audit; -use self::restore::Opened; -use crate::audit::write_audit; -use crate::{Postgres, StoreError}; - -impl Postgres { - async fn write(&self, run: RunId, record: RunRecord) -> Result<(), StoreError> { - let sequence = - i64::try_from(record.sequence().get()).map_err(|_| StoreError::Sequence(i64::MAX))?; - let mut transaction = self.pool.begin().await?; - sqlx::query!( - "INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)", - Uuid::from(run), - sequence, - serde_json::to_value(record.event())?, - ) - .execute(&mut *transaction) - .await?; - let origin = sqlx::query_scalar!( - r#"SELECT origin AS "origin!: Json" FROM runs WHERE id = $1"#, - Uuid::from(run), - ) - .fetch_one(&mut *transaction) - .await?; - write_audit(&mut *transaction, &run_audit(run, &record, &origin.0)).await?; - transaction.commit().await?; - Ok(()) - } -} - -impl RunLog for Postgres { - type Error = StoreError; - - async fn opened(&self, run: OpenedRun<'_>) -> Result<(), Self::Error> { - let number = i64::try_from(run.number.get()).map_err(|_| StoreError::Sequence(i64::MAX))?; - let first = run.jobs.first().ok_or(StoreError::EmptyRun)?; - // The run and the event that opened it are one fact. Writing them apart - // would allow a run with no history, which nothing could rebuild. - let mut transaction = self.pool.begin().await?; - sqlx::query!( - "INSERT INTO runs (id, job, number, origin, plan) VALUES ($1, $2, $3, $4, $5)", - Uuid::from(run.run), - Uuid::from(first.id()), - number, - serde_json::to_value(run.origin)?, - first.plan(), - ) - .execute(&mut *transaction) - .await?; - for (position, job) in run.jobs.iter().enumerate() { - let position = i64::try_from(position).map_err(|_| StoreError::Sequence(i64::MAX))?; - sqlx::query!( - "INSERT INTO run_jobs ( - run, id, position, key, needs, matrix, priority, requirements, secrets, plan - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", - Uuid::from(run.run), - Uuid::from(job.id()), - position, - job.key(), - serde_json::to_value(job.needs())?, - serde_json::to_value(job.matrix())?, - serde_json::to_value(job.priority())?, - serde_json::to_value(job.requirements())?, - serde_json::to_value(job.secrets())?, - job.plan(), - ) - .execute(&mut *transaction) - .await?; - } - let sequence = i64::try_from(run.record.sequence().get()) - .map_err(|_| StoreError::Sequence(i64::MAX))?; - sqlx::query!( - "INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)", - Uuid::from(run.run), - sequence, - serde_json::to_value(run.record.event())?, - ) - .execute(&mut *transaction) - .await?; - write_audit( - &mut *transaction, - &AuditEvent::new("run.triggered", "accepted") - .run(run.run) - .principal(run.origin.principal()) - .correlation(run.origin.delivery()) - .after(AuditPayload::Origin(run.origin.clone())), - ) - .await?; - let jobs = run - .jobs - .iter() - .map(|job| AuditJob { - id: job.id(), - key: job.key().to_owned(), - needs: job.needs().to_vec(), - matrix: job.matrix(), - priority: job.priority(), - requirements: job.requirements().clone(), - secrets: job.secrets().to_vec(), - }) - .collect::>(); - write_audit( - &mut *transaction, - &AuditEvent::new("workflow.compiled", "accepted") - .run(run.run) - .principal(run.origin.principal()) - .correlation(run.origin.delivery()) - .after(AuditPayload::Jobs(AuditJobs { jobs })), - ) - .await?; - transaction.commit().await?; - Ok(()) - } - - async fn appended(&self, run: RunId, record: RunRecord) -> Result<(), Self::Error> { - self.write(run, record).await - } - - /// The offset is the primary key, so a batch that arrives twice is written - /// once. A node retrying after a broken connection must not double the log. - async fn logged( - &self, - run: RunId, - job: JobId, - offset: u64, - lines: &[String], - ) -> Result<(), Self::Error> { - let offset = i64::try_from(offset).map_err(|_| StoreError::Sequence(i64::MAX))?; - sqlx::query!( - "INSERT INTO run_logs (run, job, at_offset, lines) VALUES ($1, $2, $3, $4) - ON CONFLICT (run, job, at_offset) DO NOTHING", - Uuid::from(run), - Uuid::from(job), - offset, - serde_json::to_value(lines)?, - ) - .execute(&self.pool) - .await?; - Ok(()) - } - - async fn lines(&self, run: RunId, job: JobId) -> Result, Self::Error> { - let rows = sqlx::query!( - "SELECT lines FROM run_logs WHERE run = $1 AND job = $2 ORDER BY at_offset", - Uuid::from(run), - Uuid::from(job), - ) - .fetch_all(&self.pool) - .await?; - - let mut printed = Vec::new(); - for row in rows { - let batch: Vec = serde_json::from_value(row.lines)?; - printed.extend(batch); - } - Ok(printed) - } - - async fn audited(&self, event: AuditEvent) -> Result<(), Self::Error> { - self.record_audit(event).await - } - - async fn restore(&self) -> Result, Self::Error> { - let mut jobs: BTreeMap> = BTreeMap::new(); - for row in sqlx::query!( - "SELECT run, id, key, needs, matrix, priority, requirements, secrets, plan - FROM run_jobs ORDER BY run, position", - ) - .fetch_all(&self.pool) - .await? - { - let run = RunId::from(row.run); - let priority: Priority = serde_json::from_value(row.priority)?; - let requirements: Requirements = serde_json::from_value(row.requirements)?; - let job = QueuedJob::new( - JobId::from(row.id), - row.key, - serde_json::from_value(row.needs)?, - serde_json::from_value(row.matrix)?, - row.plan, - ) - .scheduled(priority, requirements) - .referencing(serde_json::from_value(row.secrets)?); - jobs.entry(run).or_default().push(job); - } - let rows = sqlx::query!( - r#"SELECT r.id, r.number, r.origin AS "origin!: Json", e.sequence, e.event - FROM runs r - JOIN run_events e ON e.run = r.id - ORDER BY r.opened_at, r.id, e.sequence"#, - ) - .fetch_all(&self.pool) - .await?; - - let mut restored: Vec = Vec::new(); - let mut log: Vec = Vec::new(); - let mut current: Option = None; - - for row in rows { - let opened = Opened::read(row.id, row.number, row.origin.0)?; - let sequence = - u64::try_from(row.sequence).map_err(|_| StoreError::Sequence(row.sequence))?; - let event = serde_json::from_value(row.event)?; - - if current.as_ref().is_some_and(|open| open.id != opened.id) - && let Some(open) = current.take() - { - let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?; - restored.push(open.stored(definitions, std::mem::take(&mut log))); - } - current.get_or_insert(opened); - log.push(RunRecord::new(sequence.into(), event)); - } - if let Some(open) = current { - let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?; - restored.push(open.stored(definitions, log)); - } - Ok(restored) - } -} +//! How a run's history is written down and read back. + +mod audit; +mod restore; + +use std::collections::BTreeMap; + +use sqlx::types::Json; +use syncode_control_runs::{ + AuditEvent, AuditJob, AuditJobs, AuditPayload, JobId, OpenedRun, Origin, Priority, QueuedJob, + Requirements, RunEvent, RunId, RunLog, RunRecord, StoredRun, +}; +use uuid::Uuid; + +use self::audit::run_audit; +use self::restore::Opened; +use crate::audit::write_audit; +use crate::check_events::enqueue_check_snapshot; +use crate::{Postgres, StoreError}; + +impl Postgres { + async fn write(&self, run: RunId, record: RunRecord) -> Result<(), StoreError> { + let sequence = + i64::try_from(record.sequence().get()).map_err(|_| StoreError::Sequence(i64::MAX))?; + let mut transaction = self.pool.begin().await?; + sqlx::query!( + "INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)", + Uuid::from(run), + sequence, + serde_json::to_value(record.event())?, + ) + .execute(&mut *transaction) + .await?; + if affects_check_status(&record.event()) { + enqueue_check_snapshot(&mut transaction, run).await?; + } + let origin = sqlx::query_scalar!( + r#"SELECT origin AS "origin!: Json" FROM runs WHERE id = $1"#, + Uuid::from(run), + ) + .fetch_one(&mut *transaction) + .await?; + write_audit(&mut *transaction, &run_audit(run, &record, &origin.0)).await?; + transaction.commit().await?; + Ok(()) + } +} + +impl RunLog for Postgres { + type Error = StoreError; + + async fn opened(&self, run: OpenedRun<'_>) -> Result<(), Self::Error> { + let number = i64::try_from(run.number.get()).map_err(|_| StoreError::Sequence(i64::MAX))?; + let first = run.jobs.first().ok_or(StoreError::EmptyRun)?; + // The run and the event that opened it are one fact. Writing them apart + // would allow a run with no history, which nothing could rebuild. + let mut transaction = self.pool.begin().await?; + sqlx::query!( + "INSERT INTO runs (id, job, number, origin, plan) VALUES ($1, $2, $3, $4, $5)", + Uuid::from(run.run), + Uuid::from(first.id()), + number, + serde_json::to_value(run.origin)?, + first.plan(), + ) + .execute(&mut *transaction) + .await?; + for (position, job) in run.jobs.iter().enumerate() { + let position = i64::try_from(position).map_err(|_| StoreError::Sequence(i64::MAX))?; + sqlx::query!( + "INSERT INTO run_jobs ( + run, id, position, key, needs, matrix, priority, requirements, secrets, plan + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", + Uuid::from(run.run), + Uuid::from(job.id()), + position, + job.key(), + serde_json::to_value(job.needs())?, + serde_json::to_value(job.matrix())?, + serde_json::to_value(job.priority())?, + serde_json::to_value(job.requirements())?, + serde_json::to_value(job.secrets())?, + job.plan(), + ) + .execute(&mut *transaction) + .await?; + } + let sequence = i64::try_from(run.record.sequence().get()) + .map_err(|_| StoreError::Sequence(i64::MAX))?; + sqlx::query!( + "INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)", + Uuid::from(run.run), + sequence, + serde_json::to_value(run.record.event())?, + ) + .execute(&mut *transaction) + .await?; + enqueue_check_snapshot(&mut transaction, run.run).await?; + write_audit( + &mut *transaction, + &AuditEvent::new("run.triggered", "accepted") + .run(run.run) + .principal(run.origin.principal()) + .correlation(run.origin.delivery()) + .after(AuditPayload::Origin(run.origin.clone())), + ) + .await?; + let jobs = run + .jobs + .iter() + .map(|job| AuditJob { + id: job.id(), + key: job.key().to_owned(), + needs: job.needs().to_vec(), + matrix: job.matrix(), + priority: job.priority(), + requirements: job.requirements().clone(), + secrets: job.secrets().to_vec(), + }) + .collect::>(); + write_audit( + &mut *transaction, + &AuditEvent::new("workflow.compiled", "accepted") + .run(run.run) + .principal(run.origin.principal()) + .correlation(run.origin.delivery()) + .after(AuditPayload::Jobs(AuditJobs { jobs })), + ) + .await?; + transaction.commit().await?; + Ok(()) + } + + async fn appended(&self, run: RunId, record: RunRecord) -> Result<(), Self::Error> { + self.write(run, record).await + } + + /// The offset is the primary key, so a batch that arrives twice is written + /// once. A node retrying after a broken connection must not double the log. + async fn logged( + &self, + run: RunId, + job: JobId, + offset: u64, + lines: &[String], + ) -> Result<(), Self::Error> { + let offset = i64::try_from(offset).map_err(|_| StoreError::Sequence(i64::MAX))?; + sqlx::query!( + "INSERT INTO run_logs (run, job, at_offset, lines) VALUES ($1, $2, $3, $4) + ON CONFLICT (run, job, at_offset) DO NOTHING", + Uuid::from(run), + Uuid::from(job), + offset, + serde_json::to_value(lines)?, + ) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn lines(&self, run: RunId, job: JobId) -> Result, Self::Error> { + let rows = sqlx::query!( + "SELECT lines FROM run_logs WHERE run = $1 AND job = $2 ORDER BY at_offset", + Uuid::from(run), + Uuid::from(job), + ) + .fetch_all(&self.pool) + .await?; + + let mut printed = Vec::new(); + for row in rows { + let batch: Vec = serde_json::from_value(row.lines)?; + printed.extend(batch); + } + Ok(printed) + } + + async fn audited(&self, event: AuditEvent) -> Result<(), Self::Error> { + self.record_audit(event).await + } + + async fn restore(&self) -> Result, Self::Error> { + let mut jobs: BTreeMap> = BTreeMap::new(); + for row in sqlx::query!( + "SELECT run, id, key, needs, matrix, priority, requirements, secrets, plan + FROM run_jobs ORDER BY run, position", + ) + .fetch_all(&self.pool) + .await? + { + let run = RunId::from(row.run); + let priority: Priority = serde_json::from_value(row.priority)?; + let requirements: Requirements = serde_json::from_value(row.requirements)?; + let job = QueuedJob::new( + JobId::from(row.id), + row.key, + serde_json::from_value(row.needs)?, + serde_json::from_value(row.matrix)?, + row.plan, + ) + .scheduled(priority, requirements) + .referencing(serde_json::from_value(row.secrets)?); + jobs.entry(run).or_default().push(job); + } + let rows = sqlx::query!( + r#"SELECT r.id, r.number, r.origin AS "origin!: Json", e.sequence, e.event + FROM runs r + JOIN run_events e ON e.run = r.id + ORDER BY r.opened_at, r.id, e.sequence"#, + ) + .fetch_all(&self.pool) + .await?; + + let mut restored: Vec = Vec::new(); + let mut log: Vec = Vec::new(); + let mut current: Option = None; + + for row in rows { + let opened = Opened::read(row.id, row.number, row.origin.0)?; + let sequence = + u64::try_from(row.sequence).map_err(|_| StoreError::Sequence(row.sequence))?; + let event = serde_json::from_value(row.event)?; + + if current.as_ref().is_some_and(|open| open.id != opened.id) + && let Some(open) = current.take() + { + let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?; + restored.push(open.stored(definitions, std::mem::take(&mut log))); + } + current.get_or_insert(opened); + log.push(RunRecord::new(sequence.into(), event)); + } + if let Some(open) = current { + let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?; + restored.push(open.stored(definitions, log)); + } + Ok(restored) + } +} + +const fn affects_check_status(event: &RunEvent) -> bool { + matches!( + event, + RunEvent::Assigned { .. } + | RunEvent::Started { .. } + | RunEvent::Finished { .. } + | RunEvent::LeaseExpired { .. } + ) +} diff --git a/crates/control-runs/src/registry/mod.rs b/crates/control-runs/src/registry/mod.rs --- a/crates/control-runs/src/registry/mod.rs +++ b/crates/control-runs/src/registry/mod.rs @@ -1,49 +1,50 @@ -mod cancellation; -mod diagnosis; -mod error; -mod projection; -mod reporting; -mod scheduling; -mod secrets; -mod state; -mod work; - -use std::sync::{Arc, atomic::AtomicU64}; -use std::time::Duration; - -use async_lock::Mutex; -use event_listener::Event; - -use self::state::State; -use crate::{Assignment, SchedulerPolicy, SchedulingRefusal}; -pub use cancellation::{Cancellation, CancellationDispatch}; -pub use error::RunsError; - -pub const LEASE_TERM: Duration = Duration::from_secs(60); - -pub struct Runs { - pub(crate) store: Arc, - pub(crate) state: Arc>, - work: Arc, - work_version: Arc, - policy: SchedulerPolicy, -} - -#[derive(Clone, Debug)] -pub enum Dispatch { - Assigned(Box), - Empty, - Refused(Vec), -} - -impl Clone for Runs { - fn clone(&self) -> Self { - Self { - store: Arc::clone(&self.store), - state: Arc::clone(&self.state), - work: Arc::clone(&self.work), - work_version: Arc::clone(&self.work_version), - policy: self.policy, - } - } -} +mod cancellation; +mod diagnosis; +mod error; +mod projection; +mod read; +mod reporting; +mod scheduling; +mod secrets; +mod state; +mod work; + +use std::sync::{Arc, atomic::AtomicU64}; +use std::time::Duration; + +use async_lock::Mutex; +use event_listener::Event; + +use self::state::State; +use crate::{Assignment, SchedulerPolicy, SchedulingRefusal}; +pub use cancellation::{Cancellation, CancellationDispatch}; +pub use error::RunsError; + +pub const LEASE_TERM: Duration = Duration::from_secs(60); + +pub struct Runs { + pub(crate) store: Arc, + pub(crate) state: Arc>, + work: Arc, + work_version: Arc, + policy: SchedulerPolicy, +} + +#[derive(Clone, Debug)] +pub enum Dispatch { + Assigned(Box), + Empty, + Refused(Vec), +} + +impl Clone for Runs { + fn clone(&self) -> Self { + Self { + store: Arc::clone(&self.store), + state: Arc::clone(&self.state), + work: Arc::clone(&self.work), + work_version: Arc::clone(&self.work_version), + policy: self.policy, + } + } +} diff --git a/crates/control-runs/src/registry/state.rs b/crates/control-runs/src/registry/state.rs --- a/crates/control-runs/src/registry/state.rs +++ b/crates/control-runs/src/registry/state.rs @@ -1,250 +1,250 @@ -mod diagnosis; -mod scheduling; - -use std::collections::BTreeMap; - -use crate::Fence; -use crate::{ - Assignment, JobId, JobState, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError, - SchedulerPolicy, -}; -use crate::{Lease, NodeId}; - -pub struct Entry { - run: Run, - number: RunNumber, - origin: Origin, -} - -pub struct State { - entries: Vec, - organization_turns: BTreeMap, - next_turn: u64, -} - -pub(super) struct ProjectionSeed { - pub run: RunId, - pub number: RunNumber, - pub origin: Origin, - pub sequence: u64, - pub jobs: Vec, -} - -pub(super) struct ProjectionJobSeed { - pub job: JobId, - pub key: String, - pub needs: Vec, - pub state: ProjectedJobState, - pub node: Option, - pub fence: Option, -} - -impl Default for State { - fn default() -> Self { - Self { - entries: Vec::new(), - organization_turns: BTreeMap::new(), - next_turn: 1, - } - } -} - -impl State { - pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) { - self.entries.push(Entry { - run, - number, - origin, - }); - } - - pub fn next_number(&self) -> RunNumber { - self.entries - .iter() - .map(|entry| entry.number) - .max() - .map_or_else(RunNumber::first, RunNumber::after) - } - - pub fn run_by_origin(&self, origin: &Origin) -> Option { - self.entries - .iter() - .find(|entry| &entry.origin == origin) - .map(|entry| entry.run.id()) - } - - pub fn next_queued(&self) -> Option<(RunId, JobId)> { - self.entries.iter().find_map(|entry| { - entry - .run - .queued_jobs() - .next() - .map(|job| (entry.run.id(), job.id())) - }) - } - - pub fn admits_queue( - &self, - origin: &Origin, - jobs: usize, - policy: SchedulerPolicy, - ) -> Result<(), RunsError> { - let organization_open: usize = self - .entries - .iter() - .filter(|entry| entry.origin.organization() == origin.organization()) - .map(|entry| entry.run.open_jobs()) - .sum(); - if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize { - return Err(RunsError::OrganizationQueueQuota { - organization: origin.organization().to_owned(), - limit: policy.organization_queue_quota(), - }); - } - if let Some(principal) = origin.principal() { - let principal_open: usize = self - .entries - .iter() - .filter(|entry| entry.origin.principal() == Some(principal)) - .map(|entry| entry.run.open_jobs()) - .sum(); - if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize { - return Err(RunsError::PrincipalQueueQuota { - principal: principal.to_owned(), - limit: policy.principal_queue_quota(), - }); - } - } - Ok(()) - } - - pub fn assignment(&self, run: RunId, job: JobId) -> Result { - let entry = self.entry(run)?; - let definition = entry.run.job_definition(job)?; - let needs = entry.run.dependencies(job)?; - Ok(Assignment::new( - run, - job, - entry.number, - entry.origin.clone(), - definition, - needs, - )) - } - - pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> { - let organization = self.entry(run)?.origin.organization().to_owned(); - self.organization_turns.insert(organization, self.next_turn); - self.next_turn = self.next_turn.saturating_add(1); - Ok(()) - } - - pub fn runs(&self) -> impl Iterator { - self.entries.iter().map(|entry| &entry.run) - } - - pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> { - self.entries - .iter() - .flat_map(|entry| { - entry - .run - .cancellations(node) - .map(move |(job, lease)| (entry.run.id(), job, lease)) - }) - .collect() - } - - pub fn leases_held_by(&self, node: NodeId) -> usize { - self.entries - .iter() - .flat_map(|entry| entry.run.leased_jobs()) - .filter(|(_, lease)| lease.node() == node) - .count() - } - - pub fn run(&self, id: RunId) -> Result<&Run, RunsError> { - self.entry(id).map(|entry| &entry.run) - } - - pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> { - self.entries - .iter_mut() - .find(|entry| entry.run.id() == id) - .map(|entry| &mut entry.run) - .ok_or(RunsError::UnknownRun(id)) - } - - pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> { - self.entry(id).map(|entry| &entry.origin) - } - - /// Every run of one commit, unfiltered by delivery. - pub(super) fn seeds_for_commit(&self, repository: &str, commit: &str) -> Vec { - self.entries - .iter() - .filter(|entry| { - entry.origin.repository() == repository && entry.origin.commit() == commit - }) - .map(Self::projection_seed_of) - .collect() - } - - pub(super) fn projection_seeds(&self) -> Vec { - self.entries - .iter() - .filter(|entry| entry.origin.delivery().is_some()) - .map(Self::projection_seed_of) - .collect() - } - - /// The seed for one run, if it exists and is eligible for projection. Used - /// to push a single run's state synchronously (at dispatch, so the token a - /// node just received is already valid where it will be spent) alongside - /// `projection_seeds`, which the periodic sweep uses for everything else. - pub(super) fn projection_seed(&self, run: RunId) -> Option { - self.entry(run) - .ok() - .filter(|entry| entry.origin.delivery().is_some()) - .map(Self::projection_seed_of) - } - - fn projection_seed_of(entry: &Entry) -> ProjectionSeed { - ProjectionSeed { - run: entry.run.id(), - number: entry.number, - origin: entry.origin.clone(), - sequence: entry.run.sequence().get(), - jobs: entry - .run - .projected_jobs() - .map(|(job, state)| { - let (state, authority) = match state { - JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None), - JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)), - JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)), - JobState::Finished(conclusion) => { - (ProjectedJobState::Finished(conclusion), None) - } - JobState::Skipped => (ProjectedJobState::Skipped, None), - }; - ProjectionJobSeed { - job: job.id(), - key: job.key().to_owned(), - needs: job.needs().to_vec(), - state, - node: authority.map(|lease| lease.node()), - fence: authority.map(|lease| lease.fence()), - } - }) - .collect(), - } - } - - fn entry(&self, id: RunId) -> Result<&Entry, RunsError> { - self.entries - .iter() - .find(|entry| entry.run.id() == id) - .ok_or(RunsError::UnknownRun(id)) - } -} +mod diagnosis; +mod scheduling; + +use std::collections::BTreeMap; + +use crate::Fence; +use crate::{ + Assignment, JobId, JobState, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError, + SchedulerPolicy, +}; +use crate::{Lease, NodeId}; + +pub struct Entry { + pub(super) run: Run, + number: RunNumber, + pub(super) origin: Origin, +} + +pub struct State { + pub(super) entries: Vec, + organization_turns: BTreeMap, + next_turn: u64, +} + +pub(super) struct ProjectionSeed { + pub run: RunId, + pub number: RunNumber, + pub origin: Origin, + pub sequence: u64, + pub jobs: Vec, +} + +pub(super) struct ProjectionJobSeed { + pub job: JobId, + pub key: String, + pub needs: Vec, + pub state: ProjectedJobState, + pub node: Option, + pub fence: Option, +} + +impl Default for State { + fn default() -> Self { + Self { + entries: Vec::new(), + organization_turns: BTreeMap::new(), + next_turn: 1, + } + } +} + +impl State { + pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) { + self.entries.push(Entry { + run, + number, + origin, + }); + } + + pub fn next_number(&self) -> RunNumber { + self.entries + .iter() + .map(|entry| entry.number) + .max() + .map_or_else(RunNumber::first, RunNumber::after) + } + + pub fn run_by_origin(&self, origin: &Origin) -> Option { + self.entries + .iter() + .find(|entry| &entry.origin == origin) + .map(|entry| entry.run.id()) + } + + pub fn next_queued(&self) -> Option<(RunId, JobId)> { + self.entries.iter().find_map(|entry| { + entry + .run + .queued_jobs() + .next() + .map(|job| (entry.run.id(), job.id())) + }) + } + + pub fn admits_queue( + &self, + origin: &Origin, + jobs: usize, + policy: SchedulerPolicy, + ) -> Result<(), RunsError> { + let organization_open: usize = self + .entries + .iter() + .filter(|entry| entry.origin.organization() == origin.organization()) + .map(|entry| entry.run.open_jobs()) + .sum(); + if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize { + return Err(RunsError::OrganizationQueueQuota { + organization: origin.organization().to_owned(), + limit: policy.organization_queue_quota(), + }); + } + if let Some(principal) = origin.principal() { + let principal_open: usize = self + .entries + .iter() + .filter(|entry| entry.origin.principal() == Some(principal)) + .map(|entry| entry.run.open_jobs()) + .sum(); + if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize { + return Err(RunsError::PrincipalQueueQuota { + principal: principal.to_owned(), + limit: policy.principal_queue_quota(), + }); + } + } + Ok(()) + } + + pub fn assignment(&self, run: RunId, job: JobId) -> Result { + let entry = self.entry(run)?; + let definition = entry.run.job_definition(job)?; + let needs = entry.run.dependencies(job)?; + Ok(Assignment::new( + run, + job, + entry.number, + entry.origin.clone(), + definition, + needs, + )) + } + + pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> { + let organization = self.entry(run)?.origin.organization().to_owned(); + self.organization_turns.insert(organization, self.next_turn); + self.next_turn = self.next_turn.saturating_add(1); + Ok(()) + } + + pub fn runs(&self) -> impl Iterator { + self.entries.iter().map(|entry| &entry.run) + } + + pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> { + self.entries + .iter() + .flat_map(|entry| { + entry + .run + .cancellations(node) + .map(move |(job, lease)| (entry.run.id(), job, lease)) + }) + .collect() + } + + pub fn leases_held_by(&self, node: NodeId) -> usize { + self.entries + .iter() + .flat_map(|entry| entry.run.leased_jobs()) + .filter(|(_, lease)| lease.node() == node) + .count() + } + + pub fn run(&self, id: RunId) -> Result<&Run, RunsError> { + self.entry(id).map(|entry| &entry.run) + } + + pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> { + self.entries + .iter_mut() + .find(|entry| entry.run.id() == id) + .map(|entry| &mut entry.run) + .ok_or(RunsError::UnknownRun(id)) + } + + pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> { + self.entry(id).map(|entry| &entry.origin) + } + + /// Every run of one commit, unfiltered by delivery. + pub(super) fn seeds_for_commit(&self, repository: &str, commit: &str) -> Vec { + self.entries + .iter() + .filter(|entry| { + entry.origin.repository() == repository && entry.origin.commit() == commit + }) + .map(Self::projection_seed_of) + .collect() + } + + pub(super) fn projection_seeds(&self) -> Vec { + self.entries + .iter() + .filter(|entry| entry.origin.delivery().is_some()) + .map(Self::projection_seed_of) + .collect() + } + + /// The seed for one run, if it exists and is eligible for projection. Used + /// to push a single run's state synchronously (at dispatch, so the token a + /// node just received is already valid where it will be spent) alongside + /// `projection_seeds`, which the periodic sweep uses for everything else. + pub(super) fn projection_seed(&self, run: RunId) -> Option { + self.entry(run) + .ok() + .filter(|entry| entry.origin.delivery().is_some()) + .map(Self::projection_seed_of) + } + + pub(super) fn projection_seed_of(entry: &Entry) -> ProjectionSeed { + ProjectionSeed { + run: entry.run.id(), + number: entry.number, + origin: entry.origin.clone(), + sequence: entry.run.sequence().get(), + jobs: entry + .run + .projected_jobs() + .map(|(job, state)| { + let (state, authority) = match state { + JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None), + JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)), + JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)), + JobState::Finished(conclusion) => { + (ProjectedJobState::Finished(conclusion), None) + } + JobState::Skipped => (ProjectedJobState::Skipped, None), + }; + ProjectionJobSeed { + job: job.id(), + key: job.key().to_owned(), + needs: job.needs().to_vec(), + state, + node: authority.map(|lease| lease.node()), + fence: authority.map(|lease| lease.fence()), + } + }) + .collect(), + } + } + + fn entry(&self, id: RunId) -> Result<&Entry, RunsError> { + self.entries + .iter() + .find(|entry| entry.run.id() == id) + .ok_or(RunsError::UnknownRun(id)) + } +} diff --git a/crates/control-runs/src/run/access.rs b/crates/control-runs/src/run/access.rs --- a/crates/control-runs/src/run/access.rs +++ b/crates/control-runs/src/run/access.rs @@ -1,212 +1,212 @@ -use crate::job::Job; -use crate::{Dependency, Fence, JobId, JobState, Lease, QueuedJob, RunId, RunRecord, Sequence}; -use std::collections::BTreeMap; - -use super::{Conclusion, Run, RunError}; - -impl Run { - pub(super) fn advance(&mut self) { - loop { - let mut changes = Vec::new(); - for job in self - .jobs - .iter() - .filter(|job| job.state() == JobState::Waiting) - { - let dependencies: Vec<&Job> = job - .needs() - .iter() - .flat_map(|key| { - self.jobs - .iter() - .filter(move |candidate| candidate.key() == key) - }) - .collect(); - if dependencies.iter().all(|dependency| { - matches!( - dependency.state(), - JobState::Finished(_) | JobState::Skipped - ) - }) { - changes.push((job.id(), JobState::Queued)); - } - } - if changes.is_empty() { - return; - } - for (job, state) in changes { - if let Ok(job) = self.job_mut(job) { - job.set_state(state); - } - } - } - } - - pub(super) fn job(&self, id: JobId) -> Result<&Job, RunError> { - self.jobs - .iter() - .find(|job| job.id() == id) - .ok_or(RunError::UnknownJob { - run: self.id, - job: id, - }) - } - - pub(super) fn job_mut(&mut self, id: JobId) -> Result<&mut Job, RunError> { - self.jobs - .iter_mut() - .find(|job| job.id() == id) - .ok_or(RunError::UnknownJob { - run: self.id, - job: id, - }) - } - - pub(crate) fn queued_jobs(&self) -> impl Iterator { - self.jobs.iter().filter_map(|job| { - if job.state() != JobState::Queued { - return None; - } - let Some(limit) = job.matrix().max_parallel() else { - return Some(job.definition()); - }; - let active = self - .jobs - .iter() - .filter(|candidate| { - candidate.key() == job.key() - && matches!( - candidate.state(), - JobState::Assigned(_) | JobState::Running(_) - ) - }) - .count() as u64; - (active < limit).then(|| job.definition()) - }) - } - - pub(crate) fn active_jobs(&self) -> usize { - self.jobs - .iter() - .filter(|job| matches!(job.state(), JobState::Assigned(_) | JobState::Running(_))) - .count() - } - - pub(crate) fn open_jobs(&self) -> usize { - self.jobs - .iter() - .filter(|job| { - matches!( - job.state(), - JobState::Waiting - | JobState::Queued - | JobState::Assigned(_) - | JobState::Running(_) - ) - }) - .count() - } - - pub(crate) fn dependencies(&self, job: JobId) -> Result, RunError> { - let job = self.job(job)?; - job.needs() - .iter() - .map(|key| { - let states: Vec = self - .jobs - .iter() - .filter(|candidate| candidate.key() == key) - .map(Job::state) - .collect(); - let conclusion = if states - .iter() - .any(|state| matches!(state, JobState::Finished(Conclusion::Failure))) - { - Conclusion::Failure - } else if states - .iter() - .any(|state| matches!(state, JobState::Finished(Conclusion::Cancelled))) - { - Conclusion::Cancelled - } else if states.iter().any(|state| { - matches!( - state, - JobState::Finished(Conclusion::Skipped) | JobState::Skipped - ) - }) { - Conclusion::Skipped - } else { - Conclusion::Success - }; - let outputs = self - .jobs - .iter() - .filter(|candidate| candidate.key() == key) - .flat_map(|candidate| candidate.outputs().clone()) - .collect(); - Ok(Dependency::new(key.clone(), conclusion, outputs)) - }) - .collect() - } - - pub(crate) fn jobs(&self) -> impl Iterator { - self.jobs.iter().map(Job::definition) - } - - pub(crate) fn projected_jobs(&self) -> impl Iterator { - self.jobs.iter().map(|job| (job.definition(), job.state())) - } - - pub(crate) fn job_definition(&self, job: JobId) -> Result<&QueuedJob, RunError> { - self.job(job).map(Job::definition) - } - - pub fn job_state(&self, job: JobId) -> Result { - self.job(job).map(Job::state) - } - - pub(crate) fn repeats_finish( - &self, - job: JobId, - fence: Fence, - conclusion: Conclusion, - outputs: &BTreeMap, - ) -> Result { - let job = self.job(job)?; - Ok(job.fence() == fence - && job.state() == JobState::Finished(conclusion) - && job.outputs() == outputs) - } - - pub fn lease(&self, job: JobId) -> Result, RunError> { - Ok(match self.job(job)?.state() { - JobState::Assigned(lease) | JobState::Running(lease) => Some(lease), - _ => None, - }) - } - - pub(crate) fn leased_jobs(&self) -> impl Iterator + '_ { - self.jobs.iter().filter_map(|job| match job.state() { - JobState::Assigned(lease) | JobState::Running(lease) => Some((job.id(), lease)), - _ => None, - }) - } - - pub(crate) fn single_job(&self) -> Option { - (self.jobs.len() == 1).then(|| self.jobs[0].id()) - } - - pub const fn id(&self) -> RunId { - self.id - } - - pub fn sequence(&self) -> Sequence { - self.log - .last() - .map_or_else(Sequence::default, RunRecord::sequence) - } - - pub fn events(&self) -> impl Iterator { - self.log.iter().cloned() - } -} +use crate::job::Job; +use crate::{Dependency, Fence, JobId, JobState, Lease, QueuedJob, RunId, RunRecord, Sequence}; +use std::collections::BTreeMap; + +use super::{Conclusion, Run, RunError}; + +impl Run { + pub(super) fn advance(&mut self) { + loop { + let mut changes = Vec::new(); + for job in self + .jobs + .iter() + .filter(|job| job.state() == JobState::Waiting) + { + let dependencies: Vec<&Job> = job + .needs() + .iter() + .flat_map(|key| { + self.jobs + .iter() + .filter(move |candidate| candidate.key() == key) + }) + .collect(); + if dependencies.iter().all(|dependency| { + matches!( + dependency.state(), + JobState::Finished(_) | JobState::Skipped + ) + }) { + changes.push((job.id(), JobState::Queued)); + } + } + if changes.is_empty() { + return; + } + for (job, state) in changes { + if let Ok(job) = self.job_mut(job) { + job.set_state(state); + } + } + } + } + + pub(super) fn job(&self, id: JobId) -> Result<&Job, RunError> { + self.jobs + .iter() + .find(|job| job.id() == id) + .ok_or(RunError::UnknownJob { + run: self.id, + job: id, + }) + } + + pub(super) fn job_mut(&mut self, id: JobId) -> Result<&mut Job, RunError> { + self.jobs + .iter_mut() + .find(|job| job.id() == id) + .ok_or(RunError::UnknownJob { + run: self.id, + job: id, + }) + } + + pub(crate) fn queued_jobs(&self) -> impl Iterator { + self.jobs.iter().filter_map(|job| { + if job.state() != JobState::Queued { + return None; + } + let Some(limit) = job.matrix().max_parallel() else { + return Some(job.definition()); + }; + let active = self + .jobs + .iter() + .filter(|candidate| { + candidate.key() == job.key() + && matches!( + candidate.state(), + JobState::Assigned(_) | JobState::Running(_) + ) + }) + .count() as u64; + (active < limit).then(|| job.definition()) + }) + } + + pub(crate) fn active_jobs(&self) -> usize { + self.jobs + .iter() + .filter(|job| matches!(job.state(), JobState::Assigned(_) | JobState::Running(_))) + .count() + } + + pub(crate) fn open_jobs(&self) -> usize { + self.jobs + .iter() + .filter(|job| { + matches!( + job.state(), + JobState::Waiting + | JobState::Queued + | JobState::Assigned(_) + | JobState::Running(_) + ) + }) + .count() + } + + pub(crate) fn dependencies(&self, job: JobId) -> Result, RunError> { + let job = self.job(job)?; + job.needs() + .iter() + .map(|key| { + let states: Vec = self + .jobs + .iter() + .filter(|candidate| candidate.key() == key) + .map(Job::state) + .collect(); + let conclusion = if states + .iter() + .any(|state| matches!(state, JobState::Finished(Conclusion::Failure))) + { + Conclusion::Failure + } else if states + .iter() + .any(|state| matches!(state, JobState::Finished(Conclusion::Cancelled))) + { + Conclusion::Cancelled + } else if states.iter().any(|state| { + matches!( + state, + JobState::Finished(Conclusion::Skipped) | JobState::Skipped + ) + }) { + Conclusion::Skipped + } else { + Conclusion::Success + }; + let outputs = self + .jobs + .iter() + .filter(|candidate| candidate.key() == key) + .flat_map(|candidate| candidate.outputs().clone()) + .collect(); + Ok(Dependency::new(key.clone(), conclusion, outputs)) + }) + .collect() + } + + pub(crate) fn jobs(&self) -> impl Iterator { + self.jobs.iter().map(Job::definition) + } + + pub fn projected_jobs(&self) -> impl Iterator { + self.jobs.iter().map(|job| (job.definition(), job.state())) + } + + pub(crate) fn job_definition(&self, job: JobId) -> Result<&QueuedJob, RunError> { + self.job(job).map(Job::definition) + } + + pub fn job_state(&self, job: JobId) -> Result { + self.job(job).map(Job::state) + } + + pub(crate) fn repeats_finish( + &self, + job: JobId, + fence: Fence, + conclusion: Conclusion, + outputs: &BTreeMap, + ) -> Result { + let job = self.job(job)?; + Ok(job.fence() == fence + && job.state() == JobState::Finished(conclusion) + && job.outputs() == outputs) + } + + pub fn lease(&self, job: JobId) -> Result, RunError> { + Ok(match self.job(job)?.state() { + JobState::Assigned(lease) | JobState::Running(lease) => Some(lease), + _ => None, + }) + } + + pub(crate) fn leased_jobs(&self) -> impl Iterator + '_ { + self.jobs.iter().filter_map(|job| match job.state() { + JobState::Assigned(lease) | JobState::Running(lease) => Some((job.id(), lease)), + _ => None, + }) + } + + pub(crate) fn single_job(&self) -> Option { + (self.jobs.len() == 1).then(|| self.jobs[0].id()) + } + + pub const fn id(&self) -> RunId { + self.id + } + + pub fn sequence(&self) -> Sequence { + self.log + .last() + .map_or_else(Sequence::default, RunRecord::sequence) + } + + pub fn events(&self) -> impl Iterator { + self.log.iter().cloned() + } +} diff --git a/crates/control-node/proto/actions.proto b/crates/control-node/proto/actions.proto --- /dev/null +++ b/crates/control-node/proto/actions.proto @@ -1,0 +1,55 @@ +syntax = "proto3"; + +package syncode.control.v1; + +service ActionsRead { + rpc ListRuns(ListRunsRequest) returns (ListRunsResponse); + rpc GetRun(GetRunRequest) returns (ActionRun); + rpc GetJobLogs(GetJobLogsRequest) returns (GetJobLogsResponse); +} + +message ListRunsRequest { + string session_token = 1; + string owner = 2; + string repository = 3; +} + +message GetRunRequest { + string session_token = 1; + string run_id = 2; +} + +message GetJobLogsRequest { + string session_token = 1; + string run_id = 2; + string job_id = 3; +} + +message ListRunsResponse { + repeated ActionRun runs = 1; +} + +message ActionRun { + string id = 1; + uint64 number = 2; + string repository_id = 3; + string commit = 4; + string reference = 5; + string event = 6; + string workflow = 7; + string state = 8; + string conclusion = 9; + repeated ActionJob jobs = 10; +} + +message ActionJob { + string id = 1; + string key = 2; + string state = 3; + string conclusion = 4; + string node_id = 5; +} + +message GetJobLogsResponse { + repeated string lines = 1; +} diff --git a/crates/control-node/proto/identity.proto b/crates/control-node/proto/identity.proto --- /dev/null +++ b/crates/control-node/proto/identity.proto @@ -1,0 +1,60 @@ +syntax = "proto3"; + +package syncode.identity.v1; + +service Identity { + rpc ValidateSession(ValidateSessionRequest) returns (ValidateSessionResponse); + rpc CheckCapability(CheckCapabilityRequest) returns (CheckCapabilityResponse); + rpc ResolveRepository(ResolveRepositoryRequest) returns (ResolveRepositoryResponse); +} + +message ResolveRepositoryRequest { + string owner = 1; + string name = 2; +} + +message ResolveRepositoryResponse { + string repository_id = 1; +} + +enum PrincipalKind { + PRINCIPAL_KIND_UNSPECIFIED = 0; + PRINCIPAL_KIND_USER = 1; + PRINCIPAL_KIND_PLATFORM_AGENT = 2; + PRINCIPAL_KIND_LOCAL_AGENT = 3; + PRINCIPAL_KIND_ACCESS_TOKEN = 4; +} + +enum ResourceKind { + RESOURCE_KIND_UNSPECIFIED = 0; + RESOURCE_KIND_ORGANIZATION = 1; + RESOURCE_KIND_REPOSITORY = 2; + RESOURCE_KIND_INSTANCE = 3; +} + +message ValidateSessionRequest { + string session_token = 1; +} + +message ValidateSessionResponse { + string principal_id = 1; + PrincipalKind principal_kind = 2; + int64 expires_at_unix = 3; + string owner_user_id = 4; + repeated string capabilities = 5; + string audience = 6; + ResourceKind resource_kind = 7; + string resource_id = 8; +} + +message CheckCapabilityRequest { + string principal_id = 1; + PrincipalKind principal_kind = 2; + ResourceKind resource_kind = 3; + string resource_id = 4; + string capability = 5; +} + +message CheckCapabilityResponse { + bool allowed = 1; +} diff --git a/crates/control-runs/src/read.rs b/crates/control-runs/src/read.rs --- /dev/null +++ b/crates/control-runs/src/read.rs @@ -1,0 +1,17 @@ +use crate::{JobId, NodeId, Origin, ProjectedJobState, RunId, RunNumber}; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RunView { + pub run: RunId, + pub number: RunNumber, + pub origin: Origin, + pub jobs: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct JobView { + pub job: JobId, + pub key: String, + pub state: ProjectedJobState, + pub node: Option, +} diff --git a/crates/control-runs/src/registry/read.rs b/crates/control-runs/src/registry/read.rs --- /dev/null +++ b/crates/control-runs/src/registry/read.rs @@ -1,0 +1,63 @@ +use crate::{JobView, RunId, RunView}; + +use super::state::{ProjectionSeed, State}; +use super::{Runs, RunsError}; + +impl State { + fn seeds_for_repository(&self, repository: &str) -> Vec { + self.entries + .iter() + .filter(|entry| entry.origin.repository() == repository) + .map(Self::projection_seed_of) + .collect() + } + + fn seed_for_run(&self, run: RunId) -> Option { + self.entries + .iter() + .find(|entry| entry.run.id() == run) + .map(Self::projection_seed_of) + } +} + +impl Runs { + pub async fn runs_for_repository(&self, repository: &str) -> Vec { + let mut runs = self + .state + .lock() + .await + .seeds_for_repository(repository) + .into_iter() + .map(view) + .collect::>(); + runs.sort_by_key(|run| std::cmp::Reverse(run.number)); + runs + } + + pub async fn run_view(&self, run: RunId) -> Result { + self.state + .lock() + .await + .seed_for_run(run) + .map(view) + .ok_or(RunsError::UnknownRun(run)) + } +} + +fn view(seed: ProjectionSeed) -> RunView { + RunView { + run: seed.run, + number: seed.number, + origin: seed.origin, + jobs: seed + .jobs + .into_iter() + .map(|job| JobView { + job: job.job, + key: job.key, + state: job.state, + node: job.node, + }) + .collect(), + } +} diff --git a/crates/control-store/src/check_events.rs b/crates/control-store/src/check_events.rs --- /dev/null +++ b/crates/control-store/src/check_events.rs @@ -1,0 +1,203 @@ +use sqlx::{Postgres as Database, Row, Transaction}; +use syncode_control_runs::{JobState, Origin, QueuedJob, Run, RunId, RunRecord}; +use uuid::Uuid; + +use crate::{Postgres, StoreError}; + +#[derive(Clone, Debug)] +pub struct PendingCheckEvent { + pub id: i64, + pub message_id: Uuid, + pub repository_id: Uuid, + pub payload: serde_json::Value, + pub attempts: i32, +} + +pub(crate) async fn enqueue_check_snapshot( + transaction: &mut Transaction<'_, Database>, + run: RunId, +) -> Result<(), StoreError> { + let Some((repository_id, payload)) = snapshot(transaction, run).await? else { + return Ok(()); + }; + sqlx::query( + "INSERT INTO check_event_outbox (message_id, repository_id, payload) VALUES ($1, $2, $3)", + ) + .bind(Uuid::new_v4()) + .bind(repository_id) + .bind(payload) + .execute(&mut **transaction) + .await?; + Ok(()) +} + +async fn snapshot( + transaction: &mut Transaction<'_, Database>, + run_id: RunId, +) -> Result, StoreError> { + let row = sqlx::query("SELECT number, origin FROM runs WHERE id = $1") + .bind(Uuid::from(run_id)) + .fetch_one(&mut **transaction) + .await?; + let number: i64 = row.try_get("number")?; + let origin: Origin = serde_json::from_value(row.try_get("origin")?)?; + let Ok(repository_id) = Uuid::parse_str(origin.repository()) else { + return Ok(None); + }; + let jobs = load_jobs(transaction, run_id).await?; + let records = load_records(transaction, run_id).await?; + let run = Run::replay(run_id, jobs, records) + .map_err(|error| StoreError::InvalidRun(error.to_string()))?; + let checks = run + .projected_jobs() + .map(|(job, state)| { + let (state, conclusion) = check_state(state); + serde_json::json!({ + "job_id": job.id().to_string(), + "job": job.key(), + "state": state, + "conclusion": conclusion, + }) + }) + .collect::>(); + Ok(Some(( + repository_id, + serde_json::json!({ + "repository_id": repository_id, + "commit": origin.commit(), + "workflow": origin.workflow(), + "run_id": run_id.to_string(), + "run_number": number, + "run_sequence": run.sequence().get(), + "checks": checks, + }), + ))) +} + +async fn load_jobs( + transaction: &mut Transaction<'_, Database>, + run: RunId, +) -> Result, StoreError> { + let rows = sqlx::query( + "SELECT id, key, needs, matrix, priority, requirements, secrets, plan \ + FROM run_jobs WHERE run = $1 ORDER BY position", + ) + .bind(Uuid::from(run)) + .fetch_all(&mut **transaction) + .await?; + rows.into_iter() + .map(|row| { + Ok(QueuedJob::new( + row.try_get::("id")?.into(), + row.try_get("key")?, + serde_json::from_value(row.try_get("needs")?)?, + serde_json::from_value(row.try_get("matrix")?)?, + row.try_get("plan")?, + ) + .scheduled( + serde_json::from_value(row.try_get("priority")?)?, + serde_json::from_value(row.try_get("requirements")?)?, + ) + .referencing(serde_json::from_value(row.try_get("secrets")?)?)) + }) + .collect() +} + +async fn load_records( + transaction: &mut Transaction<'_, Database>, + run: RunId, +) -> Result, StoreError> { + let rows = + sqlx::query("SELECT sequence, event FROM run_events WHERE run = $1 ORDER BY sequence") + .bind(Uuid::from(run)) + .fetch_all(&mut **transaction) + .await?; + rows.into_iter() + .map(|row| { + let sequence: i64 = row.try_get("sequence")?; + let sequence = u64::try_from(sequence).map_err(|_| StoreError::Sequence(sequence))?; + Ok(RunRecord::new( + sequence.into(), + serde_json::from_value(row.try_get("event")?)?, + )) + }) + .collect() +} + +fn check_state(state: JobState) -> (&'static str, Option<&'static str>) { + match state { + JobState::Waiting | JobState::Queued => ("waiting", None), + JobState::Assigned(_) => ("assigned", None), + JobState::Running(_) => ("running", None), + JobState::Finished(conclusion) => ("finished", Some(conclusion_name(conclusion))), + JobState::Skipped => ("skipped", Some("skipped")), + } +} + +const fn conclusion_name(conclusion: syncode_control_runs::Conclusion) -> &'static str { + match conclusion { + syncode_control_runs::Conclusion::Success => "success", + syncode_control_runs::Conclusion::Failure => "failure", + syncode_control_runs::Conclusion::Cancelled => "cancelled", + syncode_control_runs::Conclusion::Skipped => "skipped", + } +} + +impl Postgres { + pub async fn claim_check_events( + &self, + limit: i64, + lease: i64, + ) -> Result, StoreError> { + let rows = sqlx::query( + "UPDATE check_event_outbox SET attempts = attempts + 1, \ + available_at = now() + make_interval(secs => $2::int) WHERE id IN ( \ + SELECT id FROM check_event_outbox WHERE delivered_at IS NULL \ + AND available_at <= now() ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED \ + ) RETURNING id, message_id, repository_id, payload, attempts", + ) + .bind(limit) + .bind(lease) + .fetch_all(&self.pool) + .await?; + rows.into_iter() + .map(|row| { + Ok(PendingCheckEvent { + id: row.try_get("id")?, + message_id: row.try_get("message_id")?, + repository_id: row.try_get("repository_id")?, + payload: row.try_get("payload")?, + attempts: row.try_get("attempts")?, + }) + }) + .collect() + } + + pub async fn mark_check_event_delivered(&self, id: i64) -> Result<(), StoreError> { + sqlx::query( + "UPDATE check_event_outbox SET delivered_at = now(), last_error = NULL WHERE id = $1", + ) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + pub async fn mark_check_event_failed( + &self, + id: i64, + error: &str, + retry_in: i64, + ) -> Result<(), StoreError> { + sqlx::query( + "UPDATE check_event_outbox SET last_error = $2, \ + available_at = now() + make_interval(secs => $3::int) WHERE id = $1", + ) + .bind(id) + .bind(error) + .bind(retry_in) + .execute(&self.pool) + .await?; + Ok(()) + } +} diff --git a/src/actions_read.rs b/src/actions_read.rs --- /dev/null +++ b/src/actions_read.rs @@ -1,0 +1,237 @@ +use syncode_control_node::ActionsReadService; +use syncode_control_node::actions_wire::{ + ActionJob, ActionRun, GetJobLogsRequest, GetJobLogsResponse, GetRunRequest, ListRunsRequest, + ListRunsResponse, +}; +use syncode_control_runs::{Conclusion, JobView, ProjectedJobState, RunId, RunLog, RunView, Runs}; +use thiserror::Error; +use tonic::{Request, Response, Status}; + +#[derive(Debug, Error)] +pub enum AuthorizationError { + #[error("the session is not valid")] + Unauthenticated, + #[error("repository access is denied")] + Denied, + #[error("repository was not found")] + NotFound, + #[error("identity is unavailable: {0}")] + Unavailable(String), +} + +#[tonic::async_trait] +pub trait ActionsAuthorization: Send + Sync + 'static { + async fn repository( + &self, + session_token: &str, + owner: &str, + name: &str, + ) -> Result; + + async fn authorize( + &self, + session_token: &str, + repository_id: &str, + ) -> Result<(), AuthorizationError>; +} + +pub struct ActionsRead { + runs: Runs, + authorization: A, +} + +impl ActionsRead { + pub const fn new(runs: Runs, authorization: A) -> Self { + Self { + runs, + authorization, + } + } +} + +#[tonic::async_trait] +impl ActionsReadService for ActionsRead +where + L: RunLog + Send + Sync + 'static, + A: ActionsAuthorization, +{ + async fn list_runs( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + required(&request.session_token, "session token")?; + required(&request.owner, "repository owner")?; + required(&request.repository, "repository name")?; + let repository_id = self + .authorization + .repository(&request.session_token, &request.owner, &request.repository) + .await + .map_err(authorize_status)?; + let runs = self + .runs + .runs_for_repository(&repository_id) + .await + .into_iter() + .map(action_run) + .collect(); + Ok(Response::new(ListRunsResponse { runs })) + } + + async fn get_run( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + required(&request.session_token, "session token")?; + let run = parse_run(&request.run_id)?; + let view = self + .runs + .run_view(run) + .await + .map_err(|_| Status::not_found("run was not found"))?; + self.authorization + .authorize(&request.session_token, view.origin.repository()) + .await + .map_err(authorize_status)?; + Ok(Response::new(action_run(view))) + } + + async fn get_job_logs( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + required(&request.session_token, "session token")?; + let run = parse_run(&request.run_id)?; + let job = request + .job_id + .parse() + .map_err(|_| Status::invalid_argument("job id is invalid"))?; + let view = self + .runs + .run_view(run) + .await + .map_err(|_| Status::not_found("run was not found"))?; + if !view.jobs.iter().any(|candidate| candidate.job == job) { + return Err(Status::not_found("job was not found in the run")); + } + self.authorization + .authorize(&request.session_token, view.origin.repository()) + .await + .map_err(authorize_status)?; + let lines = self + .runs + .lines(run, job) + .await + .map_err(|error| Status::unavailable(error.to_string()))?; + Ok(Response::new(GetJobLogsResponse { lines })) + } +} + +fn required(value: &str, field: &'static str) -> Result<(), Status> { + if value.is_empty() { + Err(Status::invalid_argument(format!("{field} is required"))) + } else { + Ok(()) + } +} + +fn parse_run(value: &str) -> Result { + value + .parse() + .map_err(|_| Status::invalid_argument("run id is invalid")) +} + +fn authorize_status(error: AuthorizationError) -> Status { + match error { + AuthorizationError::Unauthenticated => Status::unauthenticated(error.to_string()), + AuthorizationError::Denied => Status::permission_denied(error.to_string()), + AuthorizationError::NotFound => Status::not_found(error.to_string()), + AuthorizationError::Unavailable(_) => Status::unavailable(error.to_string()), + } +} + +fn action_run(run: RunView) -> ActionRun { + let (state, conclusion) = run_status(&run.jobs); + ActionRun { + id: run.run.to_string(), + number: run.number.get(), + repository_id: run.origin.repository().to_owned(), + commit: run.origin.commit().to_owned(), + reference: run.origin.reference().to_owned(), + event: run.origin.event().to_owned(), + workflow: run.origin.workflow().to_owned(), + state: state.to_owned(), + conclusion: conclusion.to_owned(), + jobs: run.jobs.into_iter().map(action_job).collect(), + } +} + +fn action_job(job: JobView) -> ActionJob { + let (state, conclusion) = job_status(job.state); + ActionJob { + id: job.job.to_string(), + key: job.key, + state: state.to_owned(), + conclusion: conclusion.to_owned(), + node_id: job.node.map_or_else(String::new, |node| node.to_string()), + } +} + +fn run_status(jobs: &[JobView]) -> (&'static str, &'static str) { + if jobs + .iter() + .any(|job| matches!(job.state, ProjectedJobState::Running)) + { + return ("running", ""); + } + if jobs + .iter() + .any(|job| matches!(job.state, ProjectedJobState::Assigned)) + { + return ("assigned", ""); + } + if jobs + .iter() + .any(|job| matches!(job.state, ProjectedJobState::Waiting)) + { + return ("waiting", ""); + } + let conclusion = if jobs + .iter() + .any(|job| matches!(job.state, ProjectedJobState::Finished(Conclusion::Failure))) + { + "failure" + } else if jobs.iter().any(|job| { + matches!( + job.state, + ProjectedJobState::Finished(Conclusion::Cancelled) + ) + }) { + "cancelled" + } else if jobs.iter().all(|job| { + matches!( + job.state, + ProjectedJobState::Skipped | ProjectedJobState::Finished(Conclusion::Skipped) + ) + }) { + "skipped" + } else { + "success" + }; + ("finished", conclusion) +} + +const fn job_status(state: ProjectedJobState) -> (&'static str, &'static str) { + match state { + ProjectedJobState::Waiting => ("waiting", ""), + ProjectedJobState::Assigned => ("assigned", ""), + ProjectedJobState::Running => ("running", ""), + ProjectedJobState::Skipped => ("finished", "skipped"), + ProjectedJobState::Finished(Conclusion::Success) => ("finished", "success"), + ProjectedJobState::Finished(Conclusion::Failure) => ("finished", "failure"), + ProjectedJobState::Finished(Conclusion::Cancelled) => ("finished", "cancelled"), + ProjectedJobState::Finished(Conclusion::Skipped) => ("finished", "skipped"), + } +} diff --git a/src/actions_read_identity.rs b/src/actions_read_identity.rs --- /dev/null +++ b/src/actions_read_identity.rs @@ -1,0 +1,135 @@ +use syncode_control_node::identity_wire::identity_client::IdentityClient; +use syncode_control_node::identity_wire::{ + CheckCapabilityRequest, PrincipalKind, ResolveRepositoryRequest, ResourceKind, + ValidateSessionRequest, +}; +use tonic::metadata::{Ascii, MetadataValue}; +use tonic::transport::Channel; +use tonic::{Code, Request}; + +use crate::actions_read::{ActionsAuthorization, AuthorizationError}; + +#[derive(Clone)] +pub struct IdentityActionsAuthorization { + client: IdentityClient, + authorization: MetadataValue, +} + +impl IdentityActionsAuthorization { + pub async fn connect( + endpoint: String, + shared_secret: String, + ) -> Result { + let authorization = format!("Bearer {shared_secret}").parse().map_err(|_| { + AuthorizationError::Unavailable("identity secret is invalid".to_owned()) + })?; + let client = IdentityClient::connect(endpoint) + .await + .map_err(|error| AuthorizationError::Unavailable(error.to_string()))?; + Ok(Self { + client, + authorization, + }) + } + + async fn principal( + &self, + session_token: &str, + ) -> Result<(String, PrincipalKind), AuthorizationError> { + let response = self + .client + .clone() + .validate_session(self.request(ValidateSessionRequest { + session_token: session_token.to_owned(), + })) + .await + .map_err(map_identity_status)? + .into_inner(); + let kind = PrincipalKind::try_from(response.principal_kind).map_err(|_| { + AuthorizationError::Unavailable( + "identity returned an invalid principal kind".to_owned(), + ) + })?; + Ok((response.principal_id, kind)) + } + + async fn require_read( + &self, + principal_id: String, + principal_kind: PrincipalKind, + repository_id: &str, + ) -> Result<(), AuthorizationError> { + let allowed = self + .client + .clone() + .check_capability(self.request(CheckCapabilityRequest { + principal_id, + principal_kind: principal_kind.into(), + resource_kind: ResourceKind::Repository.into(), + resource_id: repository_id.to_owned(), + capability: "repo:read".to_owned(), + })) + .await + .map_err(map_identity_status)? + .into_inner() + .allowed; + if allowed { + Ok(()) + } else { + Err(AuthorizationError::Denied) + } + } + + fn request(&self, body: T) -> Request { + let mut request = Request::new(body); + request + .metadata_mut() + .insert("authorization", self.authorization.clone()); + request + } +} + +#[tonic::async_trait] +impl ActionsAuthorization for IdentityActionsAuthorization { + async fn repository( + &self, + session_token: &str, + owner: &str, + name: &str, + ) -> Result { + let (principal_id, principal_kind) = self.principal(session_token).await?; + let repository_id = self + .client + .clone() + .resolve_repository(self.request(ResolveRepositoryRequest { + owner: owner.to_owned(), + name: name.to_owned(), + })) + .await + .map_err(map_identity_status)? + .into_inner() + .repository_id; + self.require_read(principal_id, principal_kind, &repository_id) + .await?; + Ok(repository_id) + } + + async fn authorize( + &self, + session_token: &str, + repository_id: &str, + ) -> Result<(), AuthorizationError> { + let (principal_id, principal_kind) = self.principal(session_token).await?; + self.require_read(principal_id, principal_kind, repository_id) + .await + } +} + +fn map_identity_status(status: tonic::Status) -> AuthorizationError { + match status.code() { + Code::Unauthenticated => AuthorizationError::Unauthenticated, + Code::PermissionDenied => AuthorizationError::Denied, + Code::NotFound => AuthorizationError::NotFound, + _ => AuthorizationError::Unavailable(status.to_string()), + } +} diff --git a/src/check_events.rs b/src/check_events.rs --- /dev/null +++ b/src/check_events.rs @@ -1,0 +1,145 @@ +use std::time::Duration; + +use hmac::{Hmac, KeyInit, Mac}; +use serde::Serialize; +use sha2::Sha256; +use syncode_control_store::{PendingCheckEvent, Postgres}; +use thiserror::Error; +use uuid::Uuid; + +const MESSAGE_TYPE: &str = "check.status.changed"; + +pub struct CheckEventPublisher { + store: Postgres, + endpoint: reqwest::Url, + secret: String, + source_node_id: Uuid, + client: reqwest::Client, + interval: Duration, + batch: i64, +} + +pub struct CheckEventPublisherConfig<'a> { + pub endpoint: &'a str, + pub secret: String, + pub source_node_id: Uuid, + pub interval: Duration, + pub batch: i64, +} + +#[derive(Debug, Error)] +pub enum CheckEventPublisherError { + #[error("check event interval, batch, and signing secret must be non-empty")] + InvalidConfiguration, + #[error("check event endpoint is invalid: {0}")] + InvalidEndpoint(#[from] url::ParseError), + #[error("check event HTTP client cannot be created: {0}")] + HttpClient(#[from] reqwest::Error), + #[error("check event outbox failed: {0}")] + Store(#[from] syncode_control_store::StoreError), +} + +#[derive(Serialize)] +struct Envelope<'a> { + message_id: Uuid, + protocol_version: &'static str, + schema_version: u32, + source_node_id: Uuid, + repository_id: Uuid, + message_type: &'static str, + sequence: i64, + term: i64, + payload: &'a serde_json::Value, +} + +impl CheckEventPublisher { + pub fn new( + store: Postgres, + config: CheckEventPublisherConfig<'_>, + ) -> Result { + if config.interval.is_zero() || config.batch <= 0 || config.secret.is_empty() { + return Err(CheckEventPublisherError::InvalidConfiguration); + } + Ok(Self { + store, + endpoint: config.endpoint.parse()?, + secret: config.secret, + source_node_id: config.source_node_id, + client: reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()?, + interval: config.interval, + batch: config.batch, + }) + } + + pub async fn run(self) -> Result<(), CheckEventPublisherError> { + let mut ticker = tokio::time::interval(self.interval); + loop { + ticker.tick().await; + self.publish_once().await?; + } + } + + pub async fn publish_once(&self) -> Result<(), CheckEventPublisherError> { + let lease = i64::try_from(self.interval.as_secs().max(30)).unwrap_or(i64::MAX); + let pending = self.store.claim_check_events(self.batch, lease).await?; + for message in pending { + match self.publish(&message).await { + Ok(()) => self.store.mark_check_event_delivered(message.id).await?, + Err(error) => { + let retry = retry_delay(message.attempts); + self.store + .mark_check_event_failed(message.id, &error, retry) + .await?; + } + } + } + Ok(()) + } + + async fn publish(&self, message: &PendingCheckEvent) -> Result<(), String> { + let body = serde_json::to_vec(&Envelope { + message_id: message.message_id, + protocol_version: "1.0", + schema_version: 1, + source_node_id: self.source_node_id, + repository_id: message.repository_id, + message_type: MESSAGE_TYPE, + sequence: message.id, + term: 1, + payload: &message.payload, + }) + .map_err(|error| error.to_string())?; + let mut mac = Hmac::::new_from_slice(self.secret.as_bytes()) + .map_err(|error| error.to_string())?; + mac.update(&body); + let signature = hex::encode(mac.finalize().into_bytes()); + let response = self + .client + .post(self.endpoint.clone()) + .header("x-syncode-event", MESSAGE_TYPE) + .header("x-syncode-delivery", message.message_id.to_string()) + .header("x-syncode-signature", signature) + .header("content-type", "application/json") + .body(body) + .send() + .await + .map_err(|error| error.to_string())?; + if response.status().is_success() { + Ok(()) + } else { + Err(format!( + "check event endpoint returned HTTP {}", + response.status().as_u16() + )) + } + } +} + +fn retry_delay(attempts: i32) -> i64 { + let exponent = u32::try_from(attempts.saturating_sub(1)) + .unwrap_or(u32::MAX) + .min(8); + 1_i64.checked_shl(exponent).unwrap_or(300).min(300) +} diff --git a/tests/actions_read.rs b/tests/actions_read.rs --- /dev/null +++ b/tests/actions_read.rs @@ -1,0 +1,149 @@ +#![allow(clippy::expect_used)] + +use std::error::Error; + +use syncode_control::actions_read::{ActionsAuthorization, ActionsRead, AuthorizationError}; +use syncode_control_node::GeneratedActionsReadServer; +use syncode_control_node::actions_wire::actions_read_client::ActionsReadClient; +use syncode_control_node::actions_wire::{GetJobLogsRequest, GetRunRequest, ListRunsRequest}; +use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, Runs}; +use tokio::net::TcpListener; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::Code; +use tonic::transport::Server; + +type TestResult = Result>; + +const REPOSITORY_ID: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89214"; + +#[derive(Clone)] +struct Authorization; + +#[tonic::async_trait] +impl ActionsAuthorization for Authorization { + async fn repository( + &self, + session_token: &str, + owner: &str, + name: &str, + ) -> Result { + if session_token != "valid-session" { + return Err(AuthorizationError::Unauthenticated); + } + if owner == "syncode" && name == "control" { + Ok(REPOSITORY_ID.to_owned()) + } else { + Err(AuthorizationError::NotFound) + } + } + + async fn authorize( + &self, + session_token: &str, + repository_id: &str, + ) -> Result<(), AuthorizationError> { + if session_token != "valid-session" { + return Err(AuthorizationError::Unauthenticated); + } + if repository_id == REPOSITORY_ID { + Ok(()) + } else { + Err(AuthorizationError::Denied) + } + } +} + +async fn server(runs: Runs) -> TestResult> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}", listener.local_addr()?); + tokio::spawn(async move { + let _ = Server::builder() + .add_service(GeneratedActionsReadServer::new(ActionsRead::new( + runs, + Authorization, + ))) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await; + }); + Ok(ActionsReadClient::connect(endpoint).await?) +} + +#[tokio::test] +async fn lists_views_and_reads_logs_through_the_public_contract() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let job = JobId::fresh(); + let run = runs + .queue( + job, + Origin::new( + REPOSITORY_ID.to_owned(), + "abc123".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".syncode/workflows/ci.yml".to_owned(), + ), + b"plan".to_vec(), + ) + .await?; + let node = NodeId::fresh(); + let assignment = runs.take_next(node).await?.expect("assignment"); + runs.logged( + run, + job, + node, + assignment.fence(), + 0, + &["first".to_owned(), "second".to_owned()], + ) + .await?; + + let mut client = server(runs).await?; + let listed = client + .list_runs(ListRunsRequest { + session_token: "valid-session".to_owned(), + owner: "syncode".to_owned(), + repository: "control".to_owned(), + }) + .await? + .into_inner(); + assert_eq!(1, listed.runs.len()); + assert_eq!(run.to_string(), listed.runs[0].id); + assert_eq!("assigned", listed.runs[0].state); + + let viewed = client + .get_run(GetRunRequest { + session_token: "valid-session".to_owned(), + run_id: run.to_string(), + }) + .await? + .into_inner(); + assert_eq!("abc123", viewed.commit); + assert_eq!(job.to_string(), viewed.jobs[0].id); + + let logs = client + .get_job_logs(GetJobLogsRequest { + session_token: "valid-session".to_owned(), + run_id: run.to_string(), + job_id: job.to_string(), + }) + .await? + .into_inner(); + assert_eq!(vec!["first", "second"], logs.lines); + Ok(()) +} + +#[tokio::test] +async fn rejects_an_invalid_session() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let mut client = server(runs).await?; + let error = client + .list_runs(ListRunsRequest { + session_token: "invalid-session".to_owned(), + owner: "syncode".to_owned(), + repository: "control".to_owned(), + }) + .await + .expect_err("invalid session must be rejected"); + assert_eq!(Code::Unauthenticated, error.code()); + Ok(()) +} diff --git a/tests/check_events.rs b/tests/check_events.rs --- /dev/null +++ b/tests/check_events.rs @@ -1,0 +1,103 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] + +use std::error::Error; +use std::sync::{Arc, Mutex}; + +use axum::body::Bytes; +use axum::extract::State; +use axum::http::{HeaderMap, StatusCode}; +use axum::routing::post; +use hmac::{Hmac, KeyInit, Mac}; +use sha2::Sha256; +use sqlx::{AssertSqlSafe, Connection, Executor, PgConnection}; +use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig}; +use syncode_control_runs::{JobId, Origin, Runs}; +use syncode_control_store::Postgres; +use uuid::Uuid; + +type Recorded = Arc)>>>; +type TestResult = Result>; + +async fn record(State(recorded): State, headers: HeaderMap, body: Bytes) -> StatusCode { + recorded + .lock() + .map(|mut value| *value = Some((headers, body.to_vec()))) + .map_or(StatusCode::INTERNAL_SERVER_ERROR, |()| { + StatusCode::NO_CONTENT + }) +} + +async fn fresh() -> TestResult { + let base = std::env::var("SYNCODE_TEST_DATABASE_URL")?; + let schema = format!("check_events_{}", Uuid::new_v4().simple()); + let mut connection = PgConnection::connect(&base).await?; + connection + .execute(AssertSqlSafe(format!("CREATE SCHEMA \"{schema}\""))) + .await?; + connection.close().await?; + let separator = if base.contains('?') { '&' } else { '?' }; + Ok(format!( + "{base}{separator}options=-c%20search_path%3D{schema}" + )) +} + +#[tokio::test] +async fn check_status_change_is_delivered_and_acknowledged() -> TestResult { + let store = Postgres::connect(&fresh().await?, 2).await?; + let repository_id = Uuid::new_v4(); + let origin = Origin::new( + repository_id.to_string(), + "1111111111111111111111111111111111111111".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ); + Runs::restored(store.clone()) + .await? + .queue(JobId::fresh(), origin, br#"{"schema":1}"#.to_vec()) + .await?; + + let recorded = Recorded::default(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}/events", listener.local_addr()?); + let app = axum::Router::new() + .route("/events", post(record)) + .with_state(Arc::clone(&recorded)); + let server = tokio::spawn(async move { axum::serve(listener, app).await }); + + CheckEventPublisher::new( + store.clone(), + CheckEventPublisherConfig { + endpoint: &endpoint, + secret: "check-secret".to_owned(), + source_node_id: Uuid::new_v4(), + interval: std::time::Duration::from_secs(1), + batch: 10, + }, + )? + .publish_once() + .await?; + + let (headers, body) = recorded + .lock() + .map_err(|_| "recorded event lock was poisoned")? + .clone() + .ok_or("check event was not delivered")?; + let signature = hex::decode(headers["x-syncode-signature"].to_str()?)?; + let mut mac = Hmac::::new_from_slice(b"check-secret")?; + mac.update(&body); + mac.verify_slice(&signature)?; + let envelope: serde_json::Value = serde_json::from_slice(&body)?; + assert_eq!(envelope["protocol_version"], "1.0"); + assert_eq!(envelope["repository_id"], repository_id.to_string()); + assert_eq!(envelope["message_type"], "check.status.changed"); + assert_eq!(envelope["payload"]["checks"][0]["state"], "waiting"); + let delivered = sqlx::query_scalar::<_, bool>( + "SELECT delivered_at IS NOT NULL FROM check_event_outbox LIMIT 1", + ) + .fetch_one(store.pool()) + .await?; + assert!(delivered); + server.abort(); + Ok(()) +}