diff --git a/tests/projection_client.rs b/tests/projection_client.rs --- a/tests/projection_client.rs +++ b/tests/projection_client.rs @@ -1,125 +1,130 @@ -#![allow(clippy::expect_used)] - -use std::error::Error; - -use axum::Json; -use axum::http::StatusCode; -use axum::routing::post; -use syncode_control_node::ProjectionClient; -use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer}; -use syncode_control_node::identity_wire::{ - CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest, - GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse, - ValidateSessionRequest, ValidateSessionResponse, -}; -use syncode_control_runs::{Forgotten, JobId, Origin, Runs}; -use tokio::net::TcpListener; -use tokio::sync::mpsc; -use tokio_stream::wrappers::TcpListenerStream; -use tonic::transport::Server; -use tonic::{Request, Response, Status}; - -type TestResult = Result>; - -struct FixtureIdentity; - -#[tonic::async_trait] -impl Identity for FixtureIdentity { - async fn validate_session( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("validate_session")) - } - - async fn check_capability( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("check_capability")) - } - - async fn resolve_repository( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("resolve_repository")) - } - - async fn get_repository_coordinates( - &self, - request: Request, - ) -> Result, Status> { - if request - .metadata() - .get("authorization") - .and_then(|value| value.to_str().ok()) - != Some("Bearer shared-secret") - { - return Err(Status::unauthenticated("missing authorization")); - } - Ok(Response::new(GetRepositoryCoordinatesResponse { - owner: "syncode".to_owned(), - name: "pipelines-demo".to_owned(), - })) - } -} - -#[tokio::test] -async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult { - let identity_listener = TcpListener::bind("127.0.0.1:0").await?; - let identity_endpoint = format!("http://{}", identity_listener.local_addr()?); - tokio::spawn(async move { - let _ = Server::builder() - .add_service(IdentityServer::new(FixtureIdentity)) - .serve_with_incoming(TcpListenerStream::new(identity_listener)) - .await; - }); - - let projection_listener = TcpListener::bind("127.0.0.1:0").await?; - let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?; - let (sent, mut received) = mpsc::unbounded_channel(); - let app = axum::Router::new().route( - "/api/internal/actions/syncode/projection", - post(move |Json(body): Json| { - let sent = sent.clone(); - async move { - let _ = sent.send(body); - StatusCode::NO_CONTENT - } - }), - ); - tokio::spawn(async move { - let _ = axum::serve(projection_listener, app).await; - }); - - let runs = Runs::restored(Forgotten::default()).await?; - let run = runs - .queue( - JobId::fresh(), - Origin::new( - "76128383-1df5-4979-9b13-c048a5287e9a".to_owned(), - "commit".to_owned(), - "refs/heads/main".to_owned(), - "push".to_owned(), - ".gitea/workflows/ci.yml".to_owned(), - ) - .with_delivery(Some("delivery".to_owned())), - b"plan".to_vec(), - ) - .await?; - let projection = runs.projection(run).await?.expect("projection"); - ProjectionClient::new( - projection_base, - String::new(), - identity_endpoint, - "shared-secret".to_owned(), - )? - .send(&projection) - .await?; - - let body = received.recv().await.expect("projection body"); - assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo"); - assert_eq!(body["origin"]["native"], true); - Ok(()) -} +#![allow(clippy::expect_used)] + +use std::error::Error; + +use axum::Json; +use axum::http::StatusCode; +use axum::routing::post; +use syncode_control_node::ProjectionClient; +use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer}; +use syncode_control_node::identity_wire::{ + CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest, + GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse, + ValidateSessionRequest, ValidateSessionResponse, +}; +use syncode_control_runs::{Forgotten, JobId, Origin, Runs}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; + +type TestResult = Result>; + +struct FixtureIdentity; + +#[tonic::async_trait] +impl Identity for FixtureIdentity { + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("resolve_repository")) + } + + async fn get_repository_coordinates( + &self, + request: Request, + ) -> Result, Status> { + if request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()) + != Some("Bearer shared-secret") + { + return Err(Status::unauthenticated("missing authorization")); + } + Ok(Response::new(GetRepositoryCoordinatesResponse { + owner: "syncode".to_owned(), + name: "pipelines-demo".to_owned(), + })) + } +} + +#[tokio::test] +async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult { + let identity_listener = TcpListener::bind("127.0.0.1:0").await?; + let identity_endpoint = format!("http://{}", identity_listener.local_addr()?); + tokio::spawn(async move { + let _ = Server::builder() + .add_service(IdentityServer::new(FixtureIdentity)) + .serve_with_incoming(TcpListenerStream::new(identity_listener)) + .await; + }); + + let projection_listener = TcpListener::bind("127.0.0.1:0").await?; + let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?; + let (sent, mut received) = mpsc::unbounded_channel(); + let app = axum::Router::new().route( + "/api/internal/actions/syncode/projection", + post(move |Json(body): Json| { + let sent = sent.clone(); + async move { + let _ = sent.send(body); + StatusCode::NO_CONTENT + } + }), + ); + tokio::spawn(async move { + let _ = axum::serve(projection_listener, app).await; + }); + + let runs = Runs::restored(Forgotten::default()).await?; + let run = runs + .queue( + JobId::fresh(), + Origin::new( + "76128383-1df5-4979-9b13-c048a5287e9a".to_owned(), + "commit".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ) + .with_delivery(Some("delivery".to_owned())), + b"plan".to_vec(), + ) + .await?; + let projection = runs.projection(run).await?.expect("projection"); + let client = ProjectionClient::new( + projection_base, + String::new(), + identity_endpoint, + "shared-secret".to_owned(), + )?; + assert_eq!( + client + .repository_coordinates("76128383-1df5-4979-9b13-c048a5287e9a") + .await?, + "syncode/pipelines-demo" + ); + client.send(&projection).await?; + + let body = received.recv().await.expect("projection body"); + assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo"); + assert_eq!(body["origin"]["native"], true); + Ok(()) +} diff --git a/crates/control-node/src/declaration.rs b/crates/control-node/src/declaration.rs --- a/crates/control-node/src/declaration.rs +++ b/crates/control-node/src/declaration.rs @@ -1,77 +1,77 @@ -use syncode_control_nodes::{ - Architecture, Capabilities, CapabilityError, Capacity, ContainerRuntime, OperatingSystem, - Volume, -}; -use syncode_control_runs::{Origin, RunNumber}; -use thiserror::Error; - -use crate::wire; - -#[derive(Debug, Error)] -pub enum DeclarationError { - #[error("this control plane does not run nodes on {0}")] - UnknownArchitecture(String), - - #[error("this control plane does not run nodes on {0}")] - UnknownSystem(String), - - #[error(transparent)] - Capability(#[from] CapabilityError), -} - -pub fn capabilities(declared: wire::Capabilities) -> Result { - let architecture = match declared.architecture.as_str() { - "amd64" => Architecture::Amd64, - "arm64" => Architecture::Arm64, - other => return Err(DeclarationError::UnknownArchitecture(other.to_owned())), - }; - let operating_system = match declared.operating_system.as_str() { - "linux" => OperatingSystem::Linux, - "macos" => OperatingSystem::MacOs, - "windows" => OperatingSystem::Windows, - other => return Err(DeclarationError::UnknownSystem(other.to_owned())), - }; - Ok(Capabilities::new( - architecture, - operating_system, - ContainerRuntime::new( - declared.container_runtime, - declared.container_runtime_version, - )?, - declared.cores, - declared.memory_bytes, - declared.labels, - )?) -} - -pub fn capacity(declared: &wire::Capacity) -> Result { - let cache = declared.cache_volume_present.then(|| { - Volume::new( - declared.cache_volume_total_bytes, - declared.cache_volume_used_bytes, - ) - }); - Ok(Capacity::new( - declared.build_volume_free_bytes, - cache, - declared - .cache_volume_present - .then(|| declared.cache_volume_path.clone()), - declared.layer_store_bytes, - )? - .with_cache_contents( - declared.cached_images.clone(), - declared.cached_actions.clone(), - )) -} - -/// Where a run came from, as an assignment states it. -pub fn origin(number: RunNumber, origin: &Origin) -> wire::Origin { - wire::Origin { - number: number.get(), - repository: origin.repository().to_owned(), - commit: origin.commit().to_owned(), - reference: origin.reference().to_owned(), - event: origin.event().to_owned(), - } -} +use syncode_control_nodes::{ + Architecture, Capabilities, CapabilityError, Capacity, ContainerRuntime, OperatingSystem, + Volume, +}; +use syncode_control_runs::{Origin, RunNumber}; +use thiserror::Error; + +use crate::wire; + +#[derive(Debug, Error)] +pub enum DeclarationError { + #[error("this control plane does not run nodes on {0}")] + UnknownArchitecture(String), + + #[error("this control plane does not run nodes on {0}")] + UnknownSystem(String), + + #[error(transparent)] + Capability(#[from] CapabilityError), +} + +pub fn capabilities(declared: wire::Capabilities) -> Result { + let architecture = match declared.architecture.as_str() { + "amd64" => Architecture::Amd64, + "arm64" => Architecture::Arm64, + other => return Err(DeclarationError::UnknownArchitecture(other.to_owned())), + }; + let operating_system = match declared.operating_system.as_str() { + "linux" => OperatingSystem::Linux, + "macos" => OperatingSystem::MacOs, + "windows" => OperatingSystem::Windows, + other => return Err(DeclarationError::UnknownSystem(other.to_owned())), + }; + Ok(Capabilities::new( + architecture, + operating_system, + ContainerRuntime::new( + declared.container_runtime, + declared.container_runtime_version, + )?, + declared.cores, + declared.memory_bytes, + declared.labels, + )?) +} + +pub fn capacity(declared: &wire::Capacity) -> Result { + let cache = declared.cache_volume_present.then(|| { + Volume::new( + declared.cache_volume_total_bytes, + declared.cache_volume_used_bytes, + ) + }); + Ok(Capacity::new( + declared.build_volume_free_bytes, + cache, + declared + .cache_volume_present + .then(|| declared.cache_volume_path.clone()), + declared.layer_store_bytes, + )? + .with_cache_contents( + declared.cached_images.clone(), + declared.cached_actions.clone(), + )) +} + +/// Where a run came from, as an assignment states it. +pub fn origin(number: RunNumber, origin: &Origin, repository: String) -> wire::Origin { + wire::Origin { + number: number.get(), + repository, + commit: origin.commit().to_owned(), + reference: origin.reference().to_owned(), + event: origin.event().to_owned(), + } +} diff --git a/crates/control-node/src/projection.rs b/crates/control-node/src/projection.rs --- a/crates/control-node/src/projection.rs +++ b/crates/control-node/src/projection.rs @@ -1,177 +1,184 @@ -use serde::Serialize; -use syncode_control_runs::{ProjectedJob, ProjectedRun, RunId, RunNumber}; -use thiserror::Error; -use tonic::Request; -use tonic::metadata::{Ascii, MetadataValue}; -use tonic::transport::{Channel, Endpoint}; -use url::Url; -use uuid::Uuid; - -use crate::identity_wire::GetRepositoryCoordinatesRequest; -use crate::identity_wire::identity_client::IdentityClient; - -/// Pushes a run's projected state to the forge. Dispatch uses it to close the -/// gap between handing a node a token and the forge knowing what that token is -/// for; the periodic sweep in the binary uses it for everything after. -#[derive(Clone)] -pub struct ProjectionClient { - endpoint: Url, - token: String, - client: reqwest::Client, - identity: IdentityClient, - authorization: MetadataValue, -} - -#[derive(Debug, Error)] -pub enum ProjectionClientError { - #[error("cannot address the projection endpoint")] - Address, - #[error("cannot address the identity endpoint")] - IdentityAddress, - #[error("cannot authorize identity requests")] - IdentityAuthorization, -} - -#[derive(Debug, Error)] -pub enum ProjectionRequestError { - #[error(transparent)] - Transport(#[from] reqwest::Error), - #[error("HTTP {status}: {body}")] - Rejected { - status: reqwest::StatusCode, - body: String, - }, - #[error("identity rejected the repository projection: {0}")] - Identity(#[from] tonic::Status), - #[error("invalid repository projection origin {0:?}")] - InvalidRepository(String), -} - -#[derive(Serialize)] -struct LegacyProjection<'a> { - run: &'a RunId, - number: &'a RunNumber, - origin: LegacyOrigin<'a>, - sequence: u64, - jobs: &'a [ProjectedJob], -} - -#[derive(Serialize)] -struct LegacyOrigin<'a> { - repository: String, - commit: &'a str, - reference: &'a str, - event: &'a str, - workflow: &'a str, - delivery: Option<&'a str>, - principal: Option<&'a str>, - secrets_allowed: bool, - native: bool, -} - -impl ProjectionClient { - pub fn new( - base: Url, - token: String, - identity_endpoint: String, - identity_shared_secret: String, - ) -> Result { - let endpoint = base - .join("api/internal/actions/syncode/projection") - .map_err(|_| ProjectionClientError::Address)?; - let identity = IdentityClient::new( - Endpoint::from_shared(identity_endpoint) - .map_err(|_| ProjectionClientError::IdentityAddress)? - .connect_lazy(), - ); - let authorization = format!("Bearer {identity_shared_secret}") - .parse() - .map_err(|_| ProjectionClientError::IdentityAuthorization)?; - Ok(Self { - endpoint, - token, - client: reqwest::Client::new(), - identity, - authorization, - }) - } - - pub async fn send(&self, projection: &ProjectedRun) -> Result<(), ProjectionRequestError> { - let projection = self.for_legacy_projection(projection).await?; - let response = self - .client - .post(self.endpoint.clone()) - .header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token)) - .json(&projection) - .send() - .await?; - let status = response.status(); - if status.is_success() { - return Ok(()); - } - let body = response.text().await?; - Err(ProjectionRequestError::Rejected { status, body }) - } - - async fn for_legacy_projection<'a>( - &self, - projection: &'a ProjectedRun, - ) -> Result, ProjectionRequestError> { - let repository = projection.origin.repository(); - let native = Uuid::parse_str(repository).is_ok(); - let repository = if native { - let mut request = Request::new(GetRepositoryCoordinatesRequest { - repository_id: repository.to_owned(), - }); - request - .metadata_mut() - .insert("authorization", self.authorization.clone()); - let coordinates = self - .identity - .clone() - .get_repository_coordinates(request) - .await? - .into_inner(); - if coordinates.owner.is_empty() - || coordinates.name.is_empty() - || coordinates.owner.contains('/') - || coordinates.name.contains('/') - { - return Err(ProjectionRequestError::InvalidRepository(format!( - "{}/{}", - coordinates.owner, coordinates.name - ))); - } - format!("{}/{}", coordinates.owner, coordinates.name) - } else { - let Some((owner, name)) = repository.split_once('/') else { - return Err(ProjectionRequestError::InvalidRepository( - repository.to_owned(), - )); - }; - if owner.is_empty() || name.is_empty() || name.contains('/') { - return Err(ProjectionRequestError::InvalidRepository( - repository.to_owned(), - )); - } - repository.to_owned() - }; - Ok(LegacyProjection { - run: &projection.run, - number: &projection.number, - origin: LegacyOrigin { - repository, - commit: projection.origin.commit(), - reference: projection.origin.reference(), - event: projection.origin.event(), - workflow: projection.origin.workflow(), - delivery: projection.origin.delivery(), - principal: projection.origin.principal(), - secrets_allowed: projection.origin.secrets_allowed(), - native, - }, - sequence: projection.sequence, - jobs: &projection.jobs, - }) - } -} +use serde::Serialize; +use syncode_control_runs::{ProjectedJob, ProjectedRun, RunId, RunNumber}; +use thiserror::Error; +use tonic::Request; +use tonic::metadata::{Ascii, MetadataValue}; +use tonic::transport::{Channel, Endpoint}; +use url::Url; +use uuid::Uuid; + +use crate::identity_wire::GetRepositoryCoordinatesRequest; +use crate::identity_wire::identity_client::IdentityClient; + +/// Pushes a run's projected state to the forge. Dispatch uses it to close the +/// gap between handing a node a token and the forge knowing what that token is +/// for; the periodic sweep in the binary uses it for everything after. +#[derive(Clone)] +pub struct ProjectionClient { + endpoint: Url, + token: String, + client: reqwest::Client, + identity: IdentityClient, + authorization: MetadataValue, +} + +#[derive(Debug, Error)] +pub enum ProjectionClientError { + #[error("cannot address the projection endpoint")] + Address, + #[error("cannot address the identity endpoint")] + IdentityAddress, + #[error("cannot authorize identity requests")] + IdentityAuthorization, +} + +#[derive(Debug, Error)] +pub enum ProjectionRequestError { + #[error(transparent)] + Transport(#[from] reqwest::Error), + #[error("HTTP {status}: {body}")] + Rejected { + status: reqwest::StatusCode, + body: String, + }, + #[error("identity rejected the repository projection: {0}")] + Identity(#[from] tonic::Status), + #[error("invalid repository projection origin {0:?}")] + InvalidRepository(String), +} + +#[derive(Serialize)] +struct LegacyProjection<'a> { + run: &'a RunId, + number: &'a RunNumber, + origin: LegacyOrigin<'a>, + sequence: u64, + jobs: &'a [ProjectedJob], +} + +#[derive(Serialize)] +struct LegacyOrigin<'a> { + repository: String, + commit: &'a str, + reference: &'a str, + event: &'a str, + workflow: &'a str, + delivery: Option<&'a str>, + principal: Option<&'a str>, + secrets_allowed: bool, + native: bool, +} + +impl ProjectionClient { + pub fn new( + base: Url, + token: String, + identity_endpoint: String, + identity_shared_secret: String, + ) -> Result { + let endpoint = base + .join("api/internal/actions/syncode/projection") + .map_err(|_| ProjectionClientError::Address)?; + let identity = IdentityClient::new( + Endpoint::from_shared(identity_endpoint) + .map_err(|_| ProjectionClientError::IdentityAddress)? + .connect_lazy(), + ); + let authorization = format!("Bearer {identity_shared_secret}") + .parse() + .map_err(|_| ProjectionClientError::IdentityAuthorization)?; + Ok(Self { + endpoint, + token, + client: reqwest::Client::new(), + identity, + authorization, + }) + } + + pub async fn send(&self, projection: &ProjectedRun) -> Result<(), ProjectionRequestError> { + let projection = self.for_legacy_projection(projection).await?; + let response = self + .client + .post(self.endpoint.clone()) + .header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token)) + .json(&projection) + .send() + .await?; + let status = response.status(); + if status.is_success() { + return Ok(()); + } + let body = response.text().await?; + Err(ProjectionRequestError::Rejected { status, body }) + } + + async fn for_legacy_projection<'a>( + &self, + projection: &'a ProjectedRun, + ) -> Result, ProjectionRequestError> { + let repository = projection.origin.repository(); + let native = Uuid::parse_str(repository).is_ok(); + let repository = self.repository_coordinates(repository).await?; + Ok(LegacyProjection { + run: &projection.run, + number: &projection.number, + origin: LegacyOrigin { + repository, + commit: projection.origin.commit(), + reference: projection.origin.reference(), + event: projection.origin.event(), + workflow: projection.origin.workflow(), + delivery: projection.origin.delivery(), + principal: projection.origin.principal(), + secrets_allowed: projection.origin.secrets_allowed(), + native, + }, + sequence: projection.sequence, + jobs: &projection.jobs, + }) + } + + pub async fn repository_coordinates( + &self, + repository: &str, + ) -> Result { + if Uuid::parse_str(repository).is_ok() { + let mut request = Request::new(GetRepositoryCoordinatesRequest { + repository_id: repository.to_owned(), + }); + request + .metadata_mut() + .insert("authorization", self.authorization.clone()); + let coordinates = self + .identity + .clone() + .get_repository_coordinates(request) + .await? + .into_inner(); + if coordinates.owner.is_empty() + || coordinates.name.is_empty() + || coordinates.owner.contains('/') + || coordinates.name.contains('/') + { + return Err(ProjectionRequestError::InvalidRepository(format!( + "{}/{}", + coordinates.owner, coordinates.name + ))); + } + Ok(format!("{}/{}", coordinates.owner, coordinates.name)) + } else { + let Some((owner, name)) = repository.split_once('/') else { + return Err(ProjectionRequestError::InvalidRepository( + repository.to_owned(), + )); + }; + if owner.is_empty() || name.is_empty() || name.contains('/') { + return Err(ProjectionRequestError::InvalidRepository( + repository.to_owned(), + )); + } + Ok(repository.to_owned()) + } + } +} diff --git a/crates/control-node/src/session/dispatch.rs b/crates/control-node/src/session/dispatch.rs --- a/crates/control-node/src/session/dispatch.rs +++ b/crates/control-node/src/session/dispatch.rs @@ -1,206 +1,211 @@ -mod assignment; - -use std::time::SystemTime; - -use prost::Message; -use syncode_control_nodes::{NodeStore, NodesError, Refusal}; -use syncode_control_runs::{Dispatch, Fence, NodeId, RunId, RunLog, Runs}; -use tokio::sync::mpsc; -use tonic::Status; - -use super::Registries; -use super::state::Session; -use crate::declaration; -use crate::error::SessionError; -use crate::outbound::{DurableDelivery, Outbound, send, send_durable}; -use crate::projection::ProjectionClient; -use crate::wire::{ControlMessage, Heartbeat, Refused, control_message}; -use assignment::assignment_body; - -pub(super) async fn offer_work( - registries: Registries<'_, L, S>, - outbound: &mut Outbound, - session: &mut Session, - sender: &mpsc::Sender>, -) -> Result<(), SessionError> { - if !registries.assignments { - return Ok(()); - } - let node = holder(session.node)?; - offer_cancellations(registries, outbound, session, sender, node).await?; - session.available = session - .maximum - .saturating_sub(registries.runs.leases_held_by(node).await); - while session.available > 0 { - registries - .nodes - .session_available(node, session.available) - .await; - let candidate = match registries.nodes.scheduling_candidate(node).await { - Ok(candidate) => candidate, - Err(NodesError::Refused(Refusal::Undeclared(_))) => { - session.available = 0; - registries.nodes.session_available(node, 0).await; - return Ok(()); - } - Err(NodesError::Refused(refusal)) => { - let body = control_message::Body::Refused(Refused { - reason: refusal.to_string(), - }); - return send(sender, outbound.next(body)).await; - } - Err(error) => return Err(error.into()), - }; - let candidates = registries.nodes.available_candidates().await; - let assignment = match registries - .runs - .take_next_among(&candidate, &candidates) - .await? - { - Dispatch::Assigned(assignment) => assignment, - Dispatch::Empty => break, - Dispatch::Refused(_) => break, - }; - retire_previous_assignments(registries.nodes, node, session, &assignment).await?; - project_before_dispatch(registries.runs, registries.projection, assignment.run()).await?; - let body = assignment_body( - &assignment, - node, - registries.authority, - registries.artifact_authority, - registries.action_artifact_url, - )?; - match send_durable(registries.nodes, node, sender, outbound, body).await? { - DurableDelivery::Pending(sent) => session.track(*sent), - DurableDelivery::Acknowledged => {} - } - session.available -= 1; - } - registries - .nodes - .session_available(node, session.available) - .await; - Ok(()) -} - -async fn retire_previous_assignments( - nodes: &syncode_control_nodes::Nodes, - node: NodeId, - session: &mut Session, - assignment: &syncode_control_runs::Assignment, -) -> Result<(), SessionError> { - for pending in nodes.outbound_pending(node).await? { - let stored = - ControlMessage::decode(pending.payload.as_slice()).map_err(SessionError::Outbox)?; - let same_job = matches!( - stored.body, - Some(control_message::Body::Assignment(ref previous)) - if previous.run == assignment.run().to_string() - && previous.job == assignment.job().to_string() - && previous.fence != assignment.fence().get() - ); - if same_job { - nodes - .outbound_acknowledged(node, &pending.message_id) - .await?; - session.acknowledged(&pending.message_id); - } - } - Ok(()) -} - -pub(super) async fn assignment_is_current( - runs: &syncode_control_runs::Runs, - node: NodeId, - message: &ControlMessage, -) -> Result { - let Some(control_message::Body::Assignment(assignment)) = message.body.as_ref() else { - return Ok(true); - }; - Ok(runs - .assignment_is_current( - assignment.run.parse()?, - assignment.job.parse()?, - node, - Fence::from(assignment.fence), - SystemTime::now(), - ) - .await) -} - -async fn offer_cancellations( - registries: Registries<'_, L, S>, - outbound: &mut Outbound, - session: &mut Session, - sender: &mpsc::Sender>, - node: NodeId, -) -> Result<(), SessionError> { - for cancellation in registries.runs.pending_cancellations(node).await { - let body = control_message::Body::Cancel(crate::wire::Cancel { - run: cancellation.run().to_string(), - fence: cancellation.fence().get(), - job: cancellation.job().to_string(), - }); - if let DurableDelivery::Pending(sent) = - send_durable(registries.nodes, node, sender, outbound, body).await? - { - session.track(*sent); - } - } - Ok(()) -} - -async fn project_before_dispatch( - runs: &Runs, - projection: &ProjectionClient, - run: RunId, -) -> Result<(), SessionError> { - let projected = runs - .projection(run) - .await? - .ok_or(SessionError::MissingProjection(run))?; - projection.send(&projected).await?; - Ok(()) -} - -pub(super) async fn retry_pending( - session: &Session, - sender: &mpsc::Sender>, -) -> Result<(), SessionError> { - for message in session.pending() { - send(sender, message.clone()).await?; - } - Ok(()) -} - -pub(super) fn holder(node: Option) -> Result { - node.ok_or(SessionError::NoHello) -} - -pub(super) async fn beat( - registries: Registries<'_, L, S>, - node: NodeId, - heartbeat: &Heartbeat, -) -> Result<(), SessionError> { - let capacity = heartbeat - .capacity - .as_ref() - .map(declaration::capacity) - .transpose()?; - registries - .nodes - .beat(node, capacity, SystemTime::now()) - .await?; - for held in &heartbeat.held { - registries - .runs - .renewed_job( - held.run.parse()?, - held.job.parse()?, - node, - Fence::from(held.fence), - ) - .await?; - } - Ok(()) -} +mod assignment; + +use std::time::SystemTime; + +use prost::Message; +use syncode_control_nodes::{NodeStore, NodesError, Refusal}; +use syncode_control_runs::{Dispatch, Fence, NodeId, RunId, RunLog, Runs}; +use tokio::sync::mpsc; +use tonic::Status; + +use super::Registries; +use super::state::Session; +use crate::declaration; +use crate::error::SessionError; +use crate::outbound::{DurableDelivery, Outbound, send, send_durable}; +use crate::projection::ProjectionClient; +use crate::wire::{ControlMessage, Heartbeat, Refused, control_message}; +use assignment::assignment_body; + +pub(super) async fn offer_work( + registries: Registries<'_, L, S>, + outbound: &mut Outbound, + session: &mut Session, + sender: &mpsc::Sender>, +) -> Result<(), SessionError> { + if !registries.assignments { + return Ok(()); + } + let node = holder(session.node)?; + offer_cancellations(registries, outbound, session, sender, node).await?; + session.available = session + .maximum + .saturating_sub(registries.runs.leases_held_by(node).await); + while session.available > 0 { + registries + .nodes + .session_available(node, session.available) + .await; + let candidate = match registries.nodes.scheduling_candidate(node).await { + Ok(candidate) => candidate, + Err(NodesError::Refused(Refusal::Undeclared(_))) => { + session.available = 0; + registries.nodes.session_available(node, 0).await; + return Ok(()); + } + Err(NodesError::Refused(refusal)) => { + let body = control_message::Body::Refused(Refused { + reason: refusal.to_string(), + }); + return send(sender, outbound.next(body)).await; + } + Err(error) => return Err(error.into()), + }; + let candidates = registries.nodes.available_candidates().await; + let assignment = match registries + .runs + .take_next_among(&candidate, &candidates) + .await? + { + Dispatch::Assigned(assignment) => assignment, + Dispatch::Empty => break, + Dispatch::Refused(_) => break, + }; + retire_previous_assignments(registries.nodes, node, session, &assignment).await?; + project_before_dispatch(registries.runs, registries.projection, assignment.run()).await?; + let repository = registries + .projection + .repository_coordinates(assignment.origin().repository()) + .await?; + let body = assignment_body( + &assignment, + repository, + node, + registries.authority, + registries.artifact_authority, + registries.action_artifact_url, + )?; + match send_durable(registries.nodes, node, sender, outbound, body).await? { + DurableDelivery::Pending(sent) => session.track(*sent), + DurableDelivery::Acknowledged => {} + } + session.available -= 1; + } + registries + .nodes + .session_available(node, session.available) + .await; + Ok(()) +} + +async fn retire_previous_assignments( + nodes: &syncode_control_nodes::Nodes, + node: NodeId, + session: &mut Session, + assignment: &syncode_control_runs::Assignment, +) -> Result<(), SessionError> { + for pending in nodes.outbound_pending(node).await? { + let stored = + ControlMessage::decode(pending.payload.as_slice()).map_err(SessionError::Outbox)?; + let same_job = matches!( + stored.body, + Some(control_message::Body::Assignment(ref previous)) + if previous.run == assignment.run().to_string() + && previous.job == assignment.job().to_string() + && previous.fence != assignment.fence().get() + ); + if same_job { + nodes + .outbound_acknowledged(node, &pending.message_id) + .await?; + session.acknowledged(&pending.message_id); + } + } + Ok(()) +} + +pub(super) async fn assignment_is_current( + runs: &syncode_control_runs::Runs, + node: NodeId, + message: &ControlMessage, +) -> Result { + let Some(control_message::Body::Assignment(assignment)) = message.body.as_ref() else { + return Ok(true); + }; + Ok(runs + .assignment_is_current( + assignment.run.parse()?, + assignment.job.parse()?, + node, + Fence::from(assignment.fence), + SystemTime::now(), + ) + .await) +} + +async fn offer_cancellations( + registries: Registries<'_, L, S>, + outbound: &mut Outbound, + session: &mut Session, + sender: &mpsc::Sender>, + node: NodeId, +) -> Result<(), SessionError> { + for cancellation in registries.runs.pending_cancellations(node).await { + let body = control_message::Body::Cancel(crate::wire::Cancel { + run: cancellation.run().to_string(), + fence: cancellation.fence().get(), + job: cancellation.job().to_string(), + }); + if let DurableDelivery::Pending(sent) = + send_durable(registries.nodes, node, sender, outbound, body).await? + { + session.track(*sent); + } + } + Ok(()) +} + +async fn project_before_dispatch( + runs: &Runs, + projection: &ProjectionClient, + run: RunId, +) -> Result<(), SessionError> { + let projected = runs + .projection(run) + .await? + .ok_or(SessionError::MissingProjection(run))?; + projection.send(&projected).await?; + Ok(()) +} + +pub(super) async fn retry_pending( + session: &Session, + sender: &mpsc::Sender>, +) -> Result<(), SessionError> { + for message in session.pending() { + send(sender, message.clone()).await?; + } + Ok(()) +} + +pub(super) fn holder(node: Option) -> Result { + node.ok_or(SessionError::NoHello) +} + +pub(super) async fn beat( + registries: Registries<'_, L, S>, + node: NodeId, + heartbeat: &Heartbeat, +) -> Result<(), SessionError> { + let capacity = heartbeat + .capacity + .as_ref() + .map(declaration::capacity) + .transpose()?; + registries + .nodes + .beat(node, capacity, SystemTime::now()) + .await?; + for held in &heartbeat.held { + registries + .runs + .renewed_job( + held.run.parse()?, + held.job.parse()?, + node, + Fence::from(held.fence), + ) + .await?; + } + Ok(()) +} diff --git a/crates/control-node/src/session/dispatch/assignment.rs b/crates/control-node/src/session/dispatch/assignment.rs --- a/crates/control-node/src/session/dispatch/assignment.rs +++ b/crates/control-node/src/session/dispatch/assignment.rs @@ -1,68 +1,70 @@ -use std::time::SystemTime; - -use syncode_control_runs::NodeId; - -use crate::error::SessionError; -use crate::wire::{Dependency, JobAssignment, control_message}; -use crate::{declaration, wire}; - -pub(super) fn assignment_body( - assignment: &syncode_control_runs::Assignment, - node: NodeId, - authority: &crate::CapabilityAuthority, - artifact_authority: &crate::ArtifactTokenAuthority, - action_artifact_url: &url::Url, -) -> Result { - let secret_capability = if assignment.secrets().is_empty() { - String::new() - } else { - authority.issue( - assignment.run(), - assignment.job(), - node, - assignment.fence(), - SystemTime::now(), - )? - }; - let actions_runtime_token = artifact_authority.issue( - assignment.run(), - assignment.number(), - assignment.job(), - node, - assignment.fence(), - SystemTime::now(), - )?; - Ok(control_message::Body::Assignment(JobAssignment { - run: assignment.run().to_string(), - job: assignment.job().to_string(), - plan: assignment.plan().to_vec(), - fence: assignment.fence().get(), - origin: Some(declaration::origin( - assignment.number(), - assignment.origin(), - )), - needs: assignment - .needs() - .iter() - .map(|dependency| Dependency { - key: dependency.key().to_owned(), - conclusion: wire_conclusion(dependency.conclusion()) as i32, - outputs: dependency.outputs().clone().into_iter().collect(), - }) - .collect(), - secrets: assignment.secrets().to_vec(), - secret_capability, - actions_runtime_token: actions_runtime_token.clone(), - action_artifact_url: action_artifact_url.to_string(), - action_artifact_capability: actions_runtime_token, - })) -} - -const fn wire_conclusion(conclusion: syncode_control_runs::Conclusion) -> wire::Conclusion { - match conclusion { - syncode_control_runs::Conclusion::Success => wire::Conclusion::Success, - syncode_control_runs::Conclusion::Failure => wire::Conclusion::Failure, - syncode_control_runs::Conclusion::Cancelled => wire::Conclusion::Cancelled, - syncode_control_runs::Conclusion::Skipped => wire::Conclusion::Skipped, - } -} +use std::time::SystemTime; + +use syncode_control_runs::NodeId; + +use crate::error::SessionError; +use crate::wire::{Dependency, JobAssignment, control_message}; +use crate::{declaration, wire}; + +pub(super) fn assignment_body( + assignment: &syncode_control_runs::Assignment, + repository: String, + node: NodeId, + authority: &crate::CapabilityAuthority, + artifact_authority: &crate::ArtifactTokenAuthority, + action_artifact_url: &url::Url, +) -> Result { + let secret_capability = if assignment.secrets().is_empty() { + String::new() + } else { + authority.issue( + assignment.run(), + assignment.job(), + node, + assignment.fence(), + SystemTime::now(), + )? + }; + let actions_runtime_token = artifact_authority.issue( + assignment.run(), + assignment.number(), + assignment.job(), + node, + assignment.fence(), + SystemTime::now(), + )?; + Ok(control_message::Body::Assignment(JobAssignment { + run: assignment.run().to_string(), + job: assignment.job().to_string(), + plan: assignment.plan().to_vec(), + fence: assignment.fence().get(), + origin: Some(declaration::origin( + assignment.number(), + assignment.origin(), + repository, + )), + needs: assignment + .needs() + .iter() + .map(|dependency| Dependency { + key: dependency.key().to_owned(), + conclusion: wire_conclusion(dependency.conclusion()) as i32, + outputs: dependency.outputs().clone().into_iter().collect(), + }) + .collect(), + secrets: assignment.secrets().to_vec(), + secret_capability, + actions_runtime_token: actions_runtime_token.clone(), + action_artifact_url: action_artifact_url.to_string(), + action_artifact_capability: actions_runtime_token, + })) +} + +const fn wire_conclusion(conclusion: syncode_control_runs::Conclusion) -> wire::Conclusion { + match conclusion { + syncode_control_runs::Conclusion::Success => wire::Conclusion::Success, + syncode_control_runs::Conclusion::Failure => wire::Conclusion::Failure, + syncode_control_runs::Conclusion::Cancelled => wire::Conclusion::Cancelled, + syncode_control_runs::Conclusion::Skipped => wire::Conclusion::Skipped, + } +}