fix: project repository coordinates to runners #47
@@ -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<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, 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<serde_json::Value>| {
|
||||
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<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, 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<serde_json::Value>| {
|
||||
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(())
|
||||
}
|
||||
@@ -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<Capabilities, DeclarationError> {
|
||||
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<Capacity, DeclarationError> {
|
||||
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<Capabilities, DeclarationError> {
|
||||
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<Capacity, DeclarationError> {
|
||||
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(),
|
||||
}
|
||||
}
|
||||
@@ -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<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[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<Self, ProjectionClientError> {
|
||||
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<LegacyProjection<'a>, 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<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[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<Self, ProjectionClientError> {
|
||||
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<LegacyProjection<'a>, 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<String, ProjectionRequestError> {
|
||||
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())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> 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<S: NodeStore>(
|
||||
nodes: &syncode_control_nodes::Nodes<S>,
|
||||
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<L: RunLog>(
|
||||
runs: &syncode_control_runs::Runs<L>,
|
||||
node: NodeId,
|
||||
message: &ControlMessage,
|
||||
) -> Result<bool, SessionError> {
|
||||
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<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
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<L: RunLog>(
|
||||
runs: &Runs<L>,
|
||||
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<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
for message in session.pending() {
|
||||
send(sender, message.clone()).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn holder(node: Option<NodeId>) -> Result<NodeId, SessionError> {
|
||||
node.ok_or(SessionError::NoHello)
|
||||
}
|
||||
|
||||
pub(super) async fn beat<L: RunLog, S: NodeStore>(
|
||||
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<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> 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<S: NodeStore>(
|
||||
nodes: &syncode_control_nodes::Nodes<S>,
|
||||
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<L: RunLog>(
|
||||
runs: &syncode_control_runs::Runs<L>,
|
||||
node: NodeId,
|
||||
message: &ControlMessage,
|
||||
) -> Result<bool, SessionError> {
|
||||
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<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
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<L: RunLog>(
|
||||
runs: &Runs<L>,
|
||||
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<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
for message in session.pending() {
|
||||
send(sender, message.clone()).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn holder(node: Option<NodeId>) -> Result<NodeId, SessionError> {
|
||||
node.ok_or(SessionError::NoHello)
|
||||
}
|
||||
|
||||
pub(super) async fn beat<L: RunLog, S: NodeStore>(
|
||||
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(())
|
||||
}
|
||||
@@ -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<control_message::Body, SessionError> {
|
||||
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<control_message::Body, SessionError> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user